PHP 怎么构建数据仓库

wen PHP项目 2026-08-04 145

本文目录导读：

PHP 怎么构建数据仓库

数据仓库架构概览
数据模型设计
ETL - 数据抽取
增量数据处理
数据分区管理
数据查询优化
数据缓存策略
完整示例
调度和监控
最佳实践建议

我来详细介绍PHP构建数据仓库的方案：

数据仓库架构概览

// 数据仓库核心架构
├── 数据源层 (Data Sources)
├── 数据抽取层 (ETL)
├── 数据存储层 (Data Storage)
├── 数据服务层 (Data Service)
└── 数据展示层 (BI/报表)

数据模型设计

星型模型实现

class StarSchema {
    // 事实表
    public function createFactTable($conn, $tableName) {
        $sql = "CREATE TABLE IF NOT EXISTS {$tableName} (
            id BIGINT PRIMARY KEY AUTO_INCREMENT,
            date_id INT,
            product_id INT,
            customer_id INT,
            quantity INT,
            amount DECIMAL(10,2),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (date_id) REFERENCES dim_date(id),
            FOREIGN KEY (product_id) REFERENCES dim_product(id),
            FOREIGN KEY (customer_id) REFERENCES dim_customer(id)
        )";
        return $conn->exec($sql);
    }
    // 维度表
    public function createDimensionTable($conn, $tableName) {
        $sql = "CREATE TABLE IF NOT EXISTS {$tableName} (
            id INT PRIMARY KEY AUTO_INCREMENT,
            name VARCHAR(255),
            attributes JSON,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )";
        return $conn->exec($sql);
    }
}

ETL - 数据抽取

class ETLProcessor {
    private $sourceConn;
    private $targetConn;
    // 抽取
    public function extract($sourceTable, $startDate, $endDate) {
        $sql = "SELECT * FROM {$sourceTable} 
                WHERE created_at BETWEEN ? AND ?";
        $stmt = $this->sourceConn->prepare($sql);
        $stmt->execute([$startDate, $endDate]);
        return $stmt->fetchAll(PDO::FETCH_ASSOC);
    }
    // 转换
    public function transform($rawData, $mappingRules) {
        $transformedData = [];
        foreach ($rawData as $row) {
            $mappedRow = [];
            foreach ($mappingRules as $sourceField => $targetField) {
                $mappedRow[$targetField] = $this->applyTransformation(
                    $row[$sourceField], 
                    $targetField
                );
            }
            $transformedData[] = $mappedRow;
        }
        return $transformedData;
    }
    // 加载
    public function load($data, $targetTable) {
        $this->targetConn->beginTransaction();
        try {
            foreach ($data as $row) {
                $columns = array_keys($row);
                $placeholders = array_fill(0, count($columns), '?');
                $sql = "INSERT INTO {$targetTable} 
                        (" . implode(',', $columns) . ") 
                        VALUES (" . implode(',', $placeholders) . ")";
                $stmt = $this->targetConn->prepare($sql);
                $stmt->execute(array_values($row));
            }
            $this->targetConn->commit();
        } catch (Exception $e) {
            $this->targetConn->rollBack();
            throw $e;
        }
    }
    private function applyTransformation($value, $targetField) {
        // 数据清洗和格式化
        switch ($targetField) {
            case 'amount':
                return round($value, 2);
            case 'date':
                return date('Y-m-d', strtotime($value));
            case 'status':
                return strtoupper(trim($value));
            default:
                return $value;
        }
    }
}

增量数据处理

class IncrementalLoader {
    private $lastSyncFile = '/tmp/last_sync.txt';
    public function getLastSyncTime() {
        if (file_exists($this->lastSyncFile)) {
            return file_get_contents($this->lastSyncFile);
        }
        return date('Y-m-d H:i:s', strtotime('-1 day'));
    }
    public function updateSyncTime() {
        file_put_contents($this->lastSyncFile, date('Y-m-d H:i:s'));
    }
    public function processIncrementalData($sourceTable) {
        $lastSync = $this->getLastSyncTime();
        $sql = "SELECT * FROM {$sourceTable} 
                WHERE updated_at > ? OR created_at > ?";
        // 处理增量数据...
        $this->updateSyncTime();
    }
}

数据分区管理

class PartitionManager {
    public function createMonthlyPartition($conn, $tableName, $year, $month) {
        $startDate = "$year-{$month}-01 00:00:00";
        $endDate = date('Y-m-t', strtotime("$year-$month-01")) . ' 23:59:59';
        $sql = "ALTER TABLE {$tableName} 
                PARTITION BY RANGE (TO_DAYS(created_at)) (
                    PARTITION p_{$year}_{$month} 
                    VALUES LESS THAN (TO_DAYS('{$endDate}'))
                )";
        return $conn->exec($sql);
    }
    public function archiveOldData($conn, $tableName, $archiveDate) {
        $sql = "CREATE TABLE {$tableName}_archive AS 
                SELECT * FROM {$tableName} 
                WHERE created_at < ?";
        $stmt = $conn->prepare($sql);
        $stmt->execute([$archiveDate]);
        // 删除已归档的数据
        $sql = "DELETE FROM {$tableName} WHERE created_at < ?";
        $stmt = $conn->prepare($sql);
        $stmt->execute([$archiveDate]);
    }
}

数据查询优化

class DataWarehouseQuery {
    public function buildOLAPQuery($factTable, $dimensions, $measures) {
        $dimColumns = [];
        foreach ($dimensions as $dim) {
            $dimColumns[] = "d_{$dim}.name AS {$dim}";
        }
        $measureSql = [];
        foreach ($measures as $measure) {
            $measureSql[] = $this->getAggregateSQL($measure);
        }
        $sql = "SELECT 
                " . implode(',', $dimColumns) . ",
                " . implode(',', $measureSql) . "
                FROM {$factTable} f
                " . $this->getDimensionJoins($dimensions) . "
                GROUP BY " . implode(',', $dimColumns) . "
                WITH ROLLUP";
        return $sql;
    }
    private function getAggregateSQL($measure) {
        $validAggregates = ['sum', 'avg', 'count', 'min', 'max'];
        $agg = $measure['aggregate'];
        $field = $measure['field'];
        if (in_array($agg, $validAggregates)) {
            return "{$agg}(f.{$field}) AS {$agg}_{$field}";
        }
        return "COUNT(*) AS record_count";
    }
}

数据缓存策略

class CacheLayer {
    private $redis;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    public function getQueryResult($key) {
        $cached = $this->redis->get($key);
        return $cached ? json_decode($cached, true) : null;
    }
    public function setQueryResult($key, $data, $ttl = 3600) {
        $this->redis->setex($key, $ttl, json_encode($data));
    }
    public function invalidateCache($tableName) {
        $pattern = "dwh:{$tableName}:*";
        $keys = $this->redis->keys($pattern);
        if (!empty($keys)) {
            $this->redis->del($keys);
        }
    }
}

完整示例

class DataWarehouseService {
    private $etlProcessor;
    private $cache;
    private $db;
    public function __construct() {
        $this->db = new PDO(
            'mysql:host=localhost;dbname=dwh',
            'user',
            'password'
        );
        $this->cache = new CacheLayer();
        $this->etlProcessor = new ETLProcessor();
    }
    public function buildWarehouse() {
        // 1. 创建表结构
        $schema = new StarSchema();
        $schema->createFactTable($this->db, 'fact_sales');
        $schema->createDimensionTable($this->db, 'dim_product');
        $schema->createDimensionTable($this->db, 'dim_customer');
        $schema->createDimensionTable($this->db, 'dim_date');
        // 2. 执行ETL
        $sourceData = $this->etlProcessor->extract('source_sales', 
                                                   '2024-01-01', 
                                                   '2024-12-31');
        // 3. 数据转换
        $mappingRules = [
            'sale_date' => 'date_id',
            'product_id' => 'product_id',
            'customer_id' => 'customer_id',
            'sale_amount' => 'amount'
        ];
        $transformedData = $this->etlProcessor->transform($sourceData, $mappingRules);
        // 4. 数据加载
        $this->etlProcessor->load($transformedData, 'fact_sales');
        // 5. 清理缓存
        $this->cache->invalidateCache('fact_sales');
        // 6. 更新元数据
        $this->updateMetadata();
    }
    public function getSalesSummary($startDate, $endDate) {
        $cacheKey = "dwh:sales:{$startDate}->{$endDate}";
        $result = $this->cache->getQueryResult($cacheKey);
        if ($result === null) {
            $query = new DataWarehouseQuery();
            $sql = $query->buildOLAPQuery(
                'fact_sales',
                ['product', 'customer'],
                [
                    ['aggregate' => 'sum', 'field' => 'amount'],
                    ['aggregate' => 'count', 'field' => 'id']
                ]
            );
            $stmt = $this->db->prepare($sql);
            $stmt->execute([$startDate, $endDate]);
            $result = $stmt->fetchAll(PDO::FETCH_ASSOC);
            $this->cache->setQueryResult($cacheKey, $result, 3600);
        }
        return $result;
    }
}

调度和监控

// 定时任务配置
class Scheduler {
    public function scheduleDailyETL() {
        // 使用 cron 表达式: 0 2 * * * (每天凌晨2点执行)
        $schedule = new Schedule();
        $schedule->addJob('etl_job', function() {
            $warehouse = new DataWarehouseService();
            $warehouse->buildWarehouse();
        });
        $schedule->run();
    }
    public function monitorQueryPerformance() {
        // 监控慢查询和性能指标
        $sql = "SHOW PROCESSLIST";
        foreach ($this->db->query($sql) as $process) {
            if ($process['Time'] > 30) {
                $this->alert("Slow Query: " . $process['Info']);
            }
        }
    }
}