PHP数据仓库与ETL流程实现

ETL（抽取、转换、加载）是把数据从源系统迁移到目标系统的过程。PHP虽然不是专业的数据处理语言，但实现简单的ETL流程很方便。

先看一个完整的ETL流程实现。

```php
interface Extractor
{
public function extract(): Generator;
}

interface Transformer
{
public function transform(array $row): ?array;
}

interface Loader
{
public function load(array $data): void;
public function getStats(): array;
}

class CsvExtractor implements Extractor
{
private string $filePath;
private int $batchSize;

public function __construct(string $filePath, int $batchSize = 100)
{
$this->filePath = $filePath;
$this->batchSize = $batchSize;
}

public function extract(): Generator
{
$handle = fopen($this->filePath, 'r');
if ($handle === false) {
throw new RuntimeException("无法打开文件: {$this->filePath}");
}

$headers = fgetcsv($handle);
if ($headers === false) {
fclose($handle);
return;
}

$batch = [];
while (($row = fgetcsv($handle)) !== false) {
$batch[] = array_combine($headers, $row);

if (count($batch) >= $this->batchSize) {
yield $batch;
$batch = [];
}
}

if (!empty($batch)) {
yield $batch;
}

fclose($handle);
}
}

class DatabaseExtractor implements Extractor
{
private PDO $pdo;
private string $query;

public function __construct(PDO $pdo, string $query)
{
$this->pdo = $pdo;
$this->query = $query;
}

public function extract(): Generator
{
$stmt = $this->pdo->query($this->query);
$batch = [];

while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
$batch[] = $row;

if (count($batch) >= 100) {
yield $batch;
$batch = [];
}
}

if (!empty($batch)) {
yield $batch;
}
}
}

class DataCleaner implements Transformer
{
private array $rules;

public function __construct(array $rules = [])
{
$this->rules = $rules;
}

public function transform(array $row): ?array
{
// 去除空行
if (empty(array_filter($row))) return null;

// 去除前后空格
$row = array_map(fn($v) => is_string($v) ? trim($v) : $v, $row);

// 应用转换规则
foreach ($this->rules as $field => $rule) {
if (isset($row[$field])) {
$row[$field] = $this->applyRule($row[$field], $rule);
}
}

return $row;
}

private function applyRule(mixed $value, array|string $rule): mixed
{
if (is_string($rule)) $rule = [$rule];

foreach ($rule as $r) {
$value = match ($r) {
'int' => (int)$value,
'float' => (float)$value,
'string' => (string)$value,
'lowercase' => strtolower((string)$value),
'uppercase' => strtoupper((string)$value),
'trim' => is_string($value) ? trim($value) : $value,
'null_if_empty' => empty($value) ? null : $value,
default => $value,
};
}

return $value;
}
}

class FieldMapper implements Transformer
{
private array $mapping;

public function __construct(array $mapping)
{
$this->mapping = $mapping;
}

public function transform(array $row): ?array
{
$mapped = [];

foreach ($this->mapping as $sourceField => $targetField) {
if (isset($row[$sourceField])) {
if (is_callable($targetField)) {
$mapped[$sourceField] = $targetField($row);
} else {
$mapped[$targetField] = $row[$sourceField];
}
}
}

return $mapped;
}
}

class DatabaseLoader implements Loader
{
private PDO $pdo;
private string $table;
private int $inserted = 0;
private int $updated = 0;
private int $failed = 0;

public function __construct(PDO $pdo, string $table)
{
$this->pdo = $pdo;
$this->table = $table;
}

public function load(array $data): void
{
if (empty($data)) return;

$columns = implode(', ', array_keys($data[0]));
$placeholders = implode(', ', array_fill(0, count($data[0]), '?'));

$sql = "INSERT INTO {$this->table} ({$columns}) VALUES ({$placeholders})";
$stmt = $this->pdo->prepare($sql);

foreach ($data as $row) {
try {
$stmt->execute(array_values($row));
$this->inserted++;
} catch (Exception $e) {
$this->failed++;
error_log("ETL加载失败: {$e->getMessage()}");
}
}
}

public function getStats(): array
{
return [
'inserted' => $this->inserted,
'updated' => $this->updated,
'failed' => $this->failed,
'total' => $this->inserted + $this->updated + $this->failed,
];
}
}

class EtlPipeline
{
private Extractor $extractor;
private array $transformers = [];
private Loader $loader;
private array $metrics = [];

public function __construct(Extractor $extractor, Loader $loader)
{
$this->extractor = $extractor;
$this->loader = $loader;
}

public function addTransformer(Transformer $transformer): void
{
$this->transformers[] = $transformer;
}

public function run(): array
{
$startTime = microtime(true);
$totalRows = 0;
$transformedRows = 0;

foreach ($this->extractor->extract() as $batch) {
foreach ($batch as $row) {
$totalRows++;

$transformed = $row;
foreach ($this->transformers as $transformer) {
$transformed = $transformer->transform($transformed);
if ($transformed === null) break;
}

if ($transformed !== null) {
$this->loader->load([$transformed]);
$transformedRows++;
}
}
}

$duration = round(microtime(true) - $startTime, 2);

return [
'total_rows' => $totalRows,
'transformed_rows' => $transformedRows,
'duration_seconds' => $duration,
'rows_per_second' => $duration > 0 ? round($totalRows / $duration, 2) : 0,
'loader_stats' => $this->loader->getStats(),
];
}
}

// 使用ETL
$pdo = new PDO('mysql:host=localhost;dbname=test', 'root', '');

$csvFile = '/tmp/source.csv';
$fp = fopen($csvFile, 'w');
fputcsv($fp, ['name', 'email', 'age', 'status']);
fputcsv($fp, [' 张三 ', 'zhangsan@test.com', '28', 'active']);
fputcsv($fp, [' 李四 ', 'lisi@test.com', '35', 'active']);
fputcsv($fp, [' 王五 ', '', '', '']);
fclose($fp);

$pipeline = new EtlPipeline(
new CsvExtractor($csvFile),
new DatabaseLoader($pdo, 'users')
);

$pipeline->addTransformer(new DataCleaner([
'age' => ['int'],
'email' => ['lowercase', 'trim'],
'name' => 'trim',
]));
$pipeline->addTransformer(new FieldMapper([
'name' => 'name',
'email' => 'email',
'age' => 'age',
'status' => 'status',
]));

$result = $pipeline->run();
print_r($result);
?>
```

ETL是数据处理的基础工具。抽取阶段从各种数据源读取，转换阶段清洗和映射，加载阶段写入目标。PHP实现ETL的优势是开发速度快，适合数据量不大的场景。大数据量的ETL建议用专业的工具如Apache NiFi、Apache Airflow或Spark。