多数据源产品表整合方案咨询：ERP数据合并与数据完整性保障

Hey, great question! Handling overlapping data from two ERP systems while preserving integrity is a common challenge in data warehousing. Here's a step-by-step approach I've used successfully in similar scenarios:

1. 先给所有源数据打"系统溯源标签" #

Before merging, the first critical step is to add a source identifier to every record as you extract it from each ERP. This ensures you never lose track of where each row came from, which is essential for troubleshooting and maintaining data lineage.

For example (using SQL Server syntax):

-- 抽取本地ERP数据到 staging 临时表
SELECT 
    'ERP_ONPREM' AS SourceSystem,
    ProductId,
    ProductName,
    Price,
    LastUpdatedTime,
    -- 包含表中所有其他字段
INTO staging.tmp_Product_OnPrem
FROM dbo.Product;

-- 抽取云端ERP数据到 staging 临时表
SELECT 
    'ERP_CLOUD' AS SourceSystem,
    ProductId,
    ProductName,
    Price,
    LastUpdatedTime,
    -- 包含表中所有其他字段
INTO staging.tmp_Product_Cloud
FROM dbp.Product;

2. 按业务规则处理重叠ProductId #

How you merge overlapping records depends entirely on your business requirements. Below are the two most common scenarios:

场景1：其中一套系统为"单一权威数据源" #

如果某套ERP（比如云端系统）是业务认可的权威数据源，另一套仅作为历史参考，可以用MERGE语句优先保留权威数据，同时将重叠的次级系统数据标记为归档：

-- 创建最终的staging表，用复合主键确保唯一性
CREATE TABLE staging.Product (
    SourceSystem VARCHAR(20) NOT NULL,
    ProductId INT NOT NULL,
    ProductName VARCHAR(100) NOT NULL,
    Price DECIMAL(18,2) NOT NULL,
    LastUpdatedTime DATETIME NOT NULL,
    IsArchived BIT DEFAULT 0,
    -- 添加其他字段
    PRIMARY KEY (SourceSystem, ProductId)
);

-- 先插入次级系统（本地）的所有数据
INSERT INTO staging.Product
SELECT * FROM staging.tmp_Product_OnPrem;

-- 合并云端数据：重叠ProductId时归档本地旧数据，无重叠则直接插入云端数据
MERGE staging.Product AS target
USING staging.tmp_Product_Cloud AS source
ON target.ProductId = source.ProductId
WHEN MATCHED THEN
    UPDATE SET target.IsArchived = 1 -- 标记本地旧数据为归档状态
WHEN NOT MATCHED THEN
    INSERT (SourceSystem, ProductId, ProductName, Price, LastUpdatedTime)
    VALUES (source.SourceSystem, source.ProductId, source.ProductName, source.Price);

场景2：按字段规则合并重叠记录 #

如果需要整合两套系统的字段数据（比如取最新价格、合并描述信息），可以用FULL OUTER JOIN搭配业务规则来解决冲突：

INSERT INTO staging.Product
SELECT 
    -- 优先取云端系统作为数据来源，无则用本地
    COALESCE(cloud.SourceSystem, onprem.SourceSystem) AS SourceSystem,
    COALESCE(cloud.ProductId, onprem.ProductId) AS ProductId,
    -- 优先使用云端产品名称，无则回退到本地
    COALESCE(cloud.ProductName, onprem.ProductName) AS ProductName,
    -- 取最新更新记录中的价格
    CASE 
        WHEN cloud.LastUpdatedTime > onprem.LastUpdatedTime THEN cloud.Price 
        ELSE onprem.Price 
    END AS Price,
    -- 合并描述字段（去除多余空格）
    TRIM(CONCAT(ISNULL(onprem.Description, ''), ' ', ISNULL(cloud.Description, ''))) AS Description,
    -- 使用最新的更新时间
    GREATEST(ISNULL(onprem.LastUpdatedTime, '1900-01-01'), ISNULL(cloud.LastUpdatedTime, '1900-01-01')) AS LastUpdatedTime
FROM staging.tmp_Product_OnPrem onprem
FULL OUTER JOIN staging.tmp_Product_Cloud cloud
ON onprem.ProductId = cloud.ProductId;

3. 合并后的数据完整性校验 #

绝对不能跳过校验环节！以下是几个关键检查点：

• 记录数匹配: 统计两套源ERP表的总记录数，确认与staging表的总记录数（包括场景1中的归档记录）一致。
• 唯一性检查: 运行以下查询确保没有重复的ProductId：

  SELECT ProductId, COUNT(*) AS DuplicateCount
  FROM staging.Product
  GROUP BY ProductId
  HAVING COUNT(*) > 1;
  

• 字段一致性: 对比源表与staging表的聚合值（比如总Price之和），避免数值型字段出现错误。

4. 为后续同步建立增量规则 #

为了避免每次全量加载数据，建议实现增量抽取逻辑：

• 在staging表中添加LastSyncTimestamp字段。
• 每次仅抽取源ERP中LastUpdatedTime > LastSyncTimestamp的记录。
• 对增量数据重复上述合并逻辑，按需更新或插入记录。

内容的提问来源于stack exchange，提问作者ahmet ciftcioglu