本文目录导读：
我来详细介绍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']);
}
}
}
}
最佳实践建议
-
性能优化
- 使用批量插入代替逐条插入
- 创建适当的索引
- 使用PARTITIONING分区表
-
可靠性
- 使用事务确保数据一致性
- 实现失败重试机制
- 定期备份数据
-
监控和优化
- 记录ETL执行日志
- 监控查询性能
- 使用EXPLAIN分析复杂查询
这个方案提供了完整的数据仓库构建思路,可以根据实际需求选择合适的组件和实现方式。
