本文目录导读：
在PHP项目中整合多业务数据表到数据集市（Data Mart）是一项典型的ETL（Extract, Transform, Load） 任务，其核心目标是将分散在业务数据库（OLTP）中的、面向流程的数据，转换为面向分析（OLAP）的、维度建模的数据。
以下是一套系统化的整合方案,从架构设计、ETL流程到PHP代码实现和性能优化，逐步拆解。
第一阶段：架构与模型设计（先设计再写代码）
不要直接写SQL去连表,先定义好数据模型。
星型模型 vs 雪花模型
数据集市通常采用星型模型：
- 事实表（Fact Table）：存储度量值（金额、数量、时间）和关联维度表的外键。
- 维度表（Dimension Table）：存储描述性属性（用户信息、商品分类、时间维度）。
示例场景：整合“订单表”、“用户表”、“商品表”、“支付日志表”。
- 事实表：
dm_sales_factdate_key(关联时间维度)user_key(关联用户维度)product_key(关联商品维度)quantity(数量)amount(金额)
- 维度表：
dim_time(年、月、日、周)dim_user(用户名、等级、注册日期)dim_product(商品名、类目、品牌)
统一数据字典
在整合前,必须解决数据冲突问题：
- 同名不同义：用户表中
status表示“激活”，订单表中status表示“已完成”。 - 同义不同名：用户表
user_id，订单表buyer_uid。 - 编码不统一：性别字段，一张表存
0/1，另一张存M/F。
决策：在数据集市中，强制使用统一的编码和字段名。
第二阶段：PHP ETL 核心流程
PHP适合做离线批处理（每天/每小时拉取一次），不适合实时流式处理，以下是标准ETL代码结构：
抽取（Extract）—— 连接多数据源
<?php
// 1. 连接源数据库（可能不止一个）
$sourceDB1 = new PDO('mysql:host=source1_host;dbname=orders_db;charset=utf8mb4', 'user1', 'pass1');
$sourceDB2 = new PDO('mysql:host=source2_host;dbname=users_db;charset=utf8mb4', 'user2', 'pass2');
// 2. 增量抽取（关键！避免每次全量跑）
$lastRunTime = getLastEtlTime('sales_fact'); // 从控制表中读取上次运行时间
$sql = "SELECT o.id, o.user_id, o.product_id, o.amount, o.created_at
FROM orders o
WHERE o.updated_at > :last_run_time"; // 或使用自增ID
$stmt = $sourceDB1->prepare($sql);
$stmt->execute([':last_run_time' => $lastRunTime]);
$orders = $stmt->fetchAll(PDO::FETCH_ASSOC);
转换（Transform）—— 处理脏数据、映射维度键
这是最耗时的环节,需要在PHP内存中完成清洗和关联。
<?php
// 假设处理每条订单记录
$transformedData = [];
foreach ($orders as $order) {
// 2.1 数据清洗
$amount = floatval($order['amount']);
if ($amount <= 0) continue; // 过滤无效数据
// 2.2 用户维度退化处理（如果用户维度简单，可直接放入事实表）
// 或者：根据 user_id 查找或创建 dim_user 表记录，返回 user_key
$userKey = getOrCreateUserDimension($sourceDB2, $order['user_id']);
// 2.3 时间维度退化
$dateKey = date('Ymd', strtotime($order['created_at']));
$timeKey = getOrCreateTimeDimension($dmDB, $order['created_at']); // 确保每天在dim_time有记录
// 2.4 商品维度关联
$productKey = getOrCreateProductDimension($sourceDB1, $order['product_id']);
// 2.5 累加度量
$transformedData[] = [
'date_key' => $dateKey,
'user_key' => $userKey,
'product_key' => $productKey,
'quantity' => 1,
'amount' => $amount,
'order_id' => $order['id'], // 业务主键，用于去重
];
}
关键函数：getOrCreateUserDimension 示例：
// 维度表建议采用SCD（缓慢变化维）策略，这里简化：不存在则插入，存在则返回key
function getOrCreateUserDimension(PDO $sourceDB, int $userId): int {
// 1. 先从维度表中查找（利用缓存可以优化）
static $userCache = [];
if (isset($userCache[$userId])) {
return $userCache[$userId];
}
$sql = "SELECT user_key FROM dim_user WHERE source_user_id = :uid";
$stmt = $dmDB->prepare($sql);
$stmt->execute([':uid' => $userId]);
$row = $stmt->fetch();
if ($row) {
$userCache[$userId] = $row['user_key'];
return $row['user_key'];
}
// 2. 从源库获取用户详细信息
$stmt = $sourceDB->prepare("SELECT name, level, register_time FROM user WHERE id = :uid");
$stmt->execute([':uid' => $userId]);
$userData = $stmt->fetch();
// 3. 插入维度表
$insertSql = "INSERT INTO dim_user (source_user_id, name, level, created_at) VALUES (:sid, :name, :level, :created)";
// ... 执行插入，获取lastInsertId
$userKey = $dmDB->lastInsertId();
$userCache[$userId] = $userKey;
return $userKey;
}
加载（Load）—— 写入数据集市
注意： 不要逐条INSERT，一定要批量INSERT，PHP逐条插入每秒只有几百条，批量可以上万。
<?php
// 批量写入事实表（示例：每次1000条）
$batchSize = 1000;
$chunks = array_chunk($transformedData, $batchSize);
$dmDB->beginTransaction();
try {
$sql = "INSERT INTO dm_sales_fact (date_key, user_key, product_key, quantity, amount, order_id)
VALUES (:date_key, :user_key, :product_key, :quantity, :amount, :order_id)
ON DUPLICATE KEY UPDATE quantity = VALUES(quantity), amount = VALUES(amount)"; // 幂等更新
$stmt = $dmDB->prepare($sql);
foreach ($chunks as $chunk) {
foreach ($chunk as $row) {
$stmt->execute($row);
}
}
$dmDB->commit();
} catch (Exception $e) {
$dmDB->rollback();
// 记录日志，发告警
}
记录ETL日志（控制表）
-- 建立控制表
CREATE TABLE etl_control (
table_name VARCHAR(100) PRIMARY KEY,
last_run_time DATETIME,
status VARCHAR(10), -- 'success', 'failed'
rows_processed INT
);
每次任务结束后更新该表,供下次增量抽取使用。
第三阶段：性能优化与常见陷阱
性能瓶颈在哪？如何优化？
| 瓶颈点 | 优化方案 |
|---|---|
| 全量关联 | 使用增量抽取（基于时间戳或自增ID），只处理变化的数据。 |
| 维度表频繁查找 | 使用内存缓存，如static变量、Redis或Hash Map存储user_id -> user_key映射。 |
| 单条插入 | 使用批量INSERT + PDO::beginTransaction + ON DUPLICATE KEY UPDATE。 |
| 数据库连接 | 使用长连接（PDO::ATTR_PERSISTENT）或连接池，减少连接建立开销。 |
| PHP内存溢出 | 使用游标（Cursor） 逐行读取源库，不要一次性fetchAll十万行。 |
数据一致性怎么保证？
- 幂等加载：在事实表对业务主键（如
order_id）建立唯一索引，使用INSERT ... ON DUPLICATE KEY UPDATE。 - 事务控制：一个批次使用一个事务，
beginTransaction+commit+rollback。 - 断点续传：即便任务中断，下次启动时仍从
last_run_time开始增量拉取。
如果业务表很大（千万级）怎么办？
- 使用SQL进行重度转换：不要让PHP做复杂的关联计算，直接在源库通过SQL（如
JOIN+GROUP BY）生成宽表中间表，PHP只负责读取和插入。 - 分片并行处理：按
user_id模10分片，启动多个PHP进程分别处理，最后合并写入。
第四阶段：建议方案选择
方案A：SQL驱动（推荐给绝大多数项目）
PHP只负责调度和日志记录,把数据整合逻辑写在源数据库的存储过程或视图里。
// PHP脚本仅做调度
$pdo->exec("CALL sp_build_sales_dm('2025-03-20')");
// 业务逻辑全在存储过程 sp_build_sales_dm 中
- 优点：利用数据库引擎（MySQL/PostgreSQL）处理关联，速度远快于PHP循环；代码量少。
- 缺点：存储过程维护难，跨数据库时麻烦（如订单在MySQL，用户日志在MongoDB）。
方案B：PHP ++ ETL框架
如果业务极其复杂（多维度、多源、需清洗），可以采用：
- 组件：
ReactPHP（异步IO）+Doctrine DBAL（数据库抽象层） +Monolog（日志）。 - 框架：
Kettle / Talend（但这是Java/Python领域），PHP可考虑自己封装基于Generator的ETL管道。
方案C：业务数据湖 + 定时任务
如果公司有数据仓库（如ClickHouse、Doris），PHP只需要调用SELECT INTO OUTFILE导出CSV，或通过HTTP Stream写入即可。
一个成熟的数据集市PHP整合流程
[每个小时/每天定时任务触发]
|
v
1. 读取 etl_control 表，获取上次运行时间 (last_run_time)
2. 连接多源数据库
3. 抽取增量数据（SQL WHERE updated_at > last_run_time）
4. 转换：
a. 清洗（去空、格式化、过滤）
b. 查（或创建）维度表，获取维度键（利用缓存）
c. 计算派生度量
5. 批量插入事实表（ON DUPLICATE KEY UPDATE 实现幂等）
6. 更新 etl_control 表（记录本次结束时间、处理行数）
7. 异常情况写入 error_log，发送邮件/钉钉告警
8. 结束
核心原则：
- 宁可在SQL中多JOIN，不要在PHP中循环查N次。
- 维度表使用缓存。
- 事实表必须批量写入。
- 永远增量，避免全量。
