数仓拉链表（来源表数据大于10000）

1.1 应用场景

数仓拉链表 文档中提供的方案，使用「参数赋值」节点将来源表中的 ID 字段输出为参数，但「参数赋值」节点要求输出的参数不能超过 10000 个，若来源表数据超过 10000 条，该方案将不适用。

本文提供数据量较大场景（来源表数据超过 10000）下拉链表的实现方案。

1.2 实现思路

1）将来源表数据同步到中间表中，便于对来源表字段做操作。

2）标记中间表和拉链表共有字段，便于后续数据关联后区分字段来源。

3）将中间表与拉链表全外连接：

注：下图示例中，来源表主键字段为ID，拉链表中也包含ID字段。

可看出：

• 来源表存在 ID，拉链表无 ID，代表新增的数据（上图红色标记的数据）。

• 来源表无 ID，拉链表存在 ID ，代表删除的数据（上图紫色标记的数据）。

• 来源表ID=拉链表ID，其他字段不一致，代表更新的数据（上图黄色标记的数据）。

4）将新增、更新、删除数据标记。

5）不同类型的数据，根据不同策略输入到拉链表中（具体参考本文 2.2 节）。

任务Demo详情参见：官方demo：https://demo.finedatalink.com/ 中的定时任务：数仓拉链表（来源表数据大于10000）

2.1 场景模拟

AUTO_INCREMENT_ID 字段为拉链表的主键，为自增字段。

来源表说明：

• 更新苹果的数据。

• 新增橙子的数据。

• 删除火龙果的数据。

接下来要将来源表数据的变化，更新到拉链表中，同时满足保留历史快照的要求。

2.2 方案说明

数仓拉链表 文档中，来源表中存在更新时间戳字段，拉链表中新增数据的开始时间为来源表的时间戳字段。

但很多场景中，用户的来源库中没有更新时间戳字段，或者更新时间戳字段不准确。针对该场景，本文提供方案。

注：用户来源表中若存在更新时间戳字段，且该字段准确，可用该字段代替「新增数据的开始时间」。

设计逻辑（以本文场景的拉链表为例）：

所以 2.1 节场景中，拉链表应该：

• 更新数据：对已存在的苹果数据，END_TIME 更新为执行本文 3.2.3 节步骤的时间；将来源表中的苹果数据，新增到拉链表中，START_TIME 为执行本文 3.2.3 节步骤的时间，END_TIME 为9999-12-31。拉链表中将存在两条苹果数据，一条为失效数据，一条为最新数据。

• 新增数据：新增橙子的数据，START_TIME 为1900-01-01（由于来源表没有时间戳字段，此为自定义的时间），END_TIME 为9999-12-31

• 删除数据：更新火龙果数据，END_TIME 为执行本文 3.2.3 节步骤的时间。

3.1 创建中间表

拉链表与来源表中，共有字段名称相同；本文方案需要将来源表数据与拉链表数据进行全外连接，识别新增、删除、更新数据，数据关联前需要修改来源表和目标表字段名，便于区分字段。

所以将来源表数据同步到中间表中，后续对中间表进行操作。

1）新建定时任务，拖入「数据同步」节点，取出来源表数据。如下图所示：

2）将数据写入到中间表中，设置 ID 为中间表的主键。如下图所示：

写入方式选择「清空目标表，再写入数据」，便于后期定期执行任务。

3）右键数据同步节点，选择「运行节点」，运行该节点，生成中间表，便于后续使用。

3.2 数仓拉链表实现

3.2.1 中间表与拉链表取并集

中间表与拉链表字段后加后缀，便于后续数据关联后区分字段来源。

1）拖入「数据转换」节点，进入「数据转换」节点。

2）拖入「DB表输入」算子，读取中间表数据。如下图所示：

3）拖入「字段设置」算子，在中间表的每个字段名称后加上_CUR，方便后续数据关联后识别字段来源。如下图所示：

4）拉链表中没被删除过的数据，END_TIME 为9999-12-31，将没被删除过的数据筛选出来，与中间表关联。如下图所示：

SELECT * FROM `demotest`.`ZIPPER`
WHERE DATE_FORMAT(END_TIME, '%Y-%m-%d')='9999-12-31'

5）拖入「字段设置」算子，在拉链表的每个字段名称后加上_PRE，方便后续数据关联后识别字段来源。

其中拉链表主键字段 AUTO_INCREMENT_ID 不用加此后缀，因为中间表中没有包含此字段，该字段无需做处理便于后续区），如下图所示：

6）拖入「数据关联」算子，将中间表与拉链表数据取并集（全外连接）。如下图所示：

点击「数据预览」，如下图所示：

可看出：

• ID_CUR（来源表）存在，ID_PRE（拉链表）无，代表新增的数据（上图红色标记的数据）。

• ID_CUR（来源表）无，ID_PRE（拉链表）存在，代表删除的数据（上图紫色标记的数据）。

• ID_CUR（来源表）=ID_PRE（拉链表），其他字段不一致，代表更新的数据（上图黄色标记的数据）。

后续可新增一列，对数据进行标记。

3.2.2 标记数据

3.2.1 节中数据关联后，我们已总结新增、删除、更新数据的特征。本节，我们标记这些数据。

1）拖入「新增计算列」算子，对数据进行标记。如下图所示：

注：公式中字段为点击生成。

IF(ISNULL(ID_PRE),"新增",IF(ISNULL(ID_CUR),"删除",IF(OR(NAME_CUR<>NAME_PRE,UNIT_CUR<>UNIT_PRE),
"更新","相同")))

2）点击「数据预览」，如下图所示：

3.2.3 分发数据准备

拉链表中数据应该：

所以，需要新增数据列：

1）START_TIME：值为1900-01-01，为新增数据的开始时间

2）END_TIME：值为9999-12-31，为新增数据的结束时间；更新-新插入数据的结束时间

3）TIME：执行该算子的时间，值为now()

拖入「新增计算列」算子，新增计算列。如下图所示：

3.2.4 数据分发

1）拖入 4 个「DB表输出」算子，与前面的「新增计算列」算子相连。

2）右键点击「新增计算列1」算子，点击「数据分发」。设置如下图所示：

3）点击「DB表输出」算子，处理新增数据，将新增数据输入到拉链表中。如下图所示：

将标记为新增的数据写入到拉链表中，拉链表中数据的开始时间对应新增字段 START_TIME ，值为 1900-01-01 ；拉链表数据的结束时间对应新增字段 END_TIME ，值为 9999-12-31 。

写入方式选择「追加写入数据」。

4）点击「DB表输出1」算子，处理删除数据，更新删除数据的 END_TIME 。如下图所示：

将标记为删除的数据写入到拉链表中，拉链表数据的结束时间对应新增字段 TIME ，值为now() 

写入方式选择「插入/更新/删除数据」中的更新，标识字段为Type，标识值为删除，逻辑主键为AUTO_INCREMENT_ID。如下图所示：

5）点击「DB表输出2」算子，更新拉链表中存在的旧更新数据中 END_TIME。如下图所示：

将标记为更新的数据写入到拉链表中，拉链表数据的结束时间对应新增字段 TIME ，值为now() 

写入方式选择「插入/更新/删除数据」中的更新，标识字段为Type，标识值为更新，逻辑主键为AUTO_INCREMENT_ID。如下图所示：

6）点击「DB表输出3」算子，将来源表中的更新数据，新增到拉链表中，拉链表中的 START_TIME 对应 3.2.3 节中的 TIME，拉链表中的 END_TIME 对应 3.2.3 节中的 END_TIME。如下图所示：

写入方式选择「追加写入数据」。

7）点击右上角「保存」按钮。

3.3 效果查看

点击右上角「保存并运行」按钮，运行成功后，拉链表如下图所示：