DataArts(DGC) -> DataWorks

一、导出DataArts调度任务流

华为DataArts提供了作业、脚本和资源的批量导出能力，用户可根据本文档的指示构造一个DataArts调度信息包。LHM调度导出工具将对DataArts调度信息包进行解析，并生成调度迁移标准包。

1 前置条件

准备JDK17运行环境和华为云账号，账号需对待迁移DataArts作业、脚本和资源具有导出权限。

2 构造DataArts调度信息包

2.1 DataArts调度信息包结构

调度信息包的结构如下，包含数据开发作业（任务流）、数据开发节点脚本、资源文件、数据集成脚本。用户需在DataArts中自行分别导出部分，并完成组包：

.
├── project.json // 项目基本信息
├── cdms
│   └──<集群名称>
│       └── <cdm_***>.json // 数据集成脚本
└── jobs
    ├── jobs
    │   └── <***>.job // 数据开发作业（任务流）
    ├── resources
    │   └── <***>.resource // 资源文件
    └── scripts
        └── <***>.script // 数据开发节点脚本

2.2 构造项目基本信息

project.json中需填写项目基本信息，包括项目Id和名称。

{
  "dataArtsWorkSpaceId": "d494af814b4b496e8b1bc7a8eee68d7a",
  "dataArtsWorkspaceName": "default"
}

2.3 导出数据开发作业（工作流定义）

打开DataArts控制台，点击数据开发：https://console.huaweicloud.com/dayu

在作业开发页面，导出作业。下载后解压缩至./jobs/jobs目录下。

2.4 导出任务脚本

打开DataArts控制台，点击数据开发：https://console.huaweicloud.com/dayu

在脚本开发页面，导出脚本。下载后解压缩至./jobs/scripts目录下。

2.5 导出资源文件

打开DataArts控制台，点击数据开发：https://console.huaweicloud.com/dayu

在资源管理页面，导出资源。下载后解压缩至./jobs/resources目录下。

2.6 导出数据集成任务脚本

打开DataArts控制台，点击数据集成：https://console.huaweicloud.com/dayu

在集群管理中，导出数据集成任务脚本。下载后解压缩至./cdms目录下。

3 运行调度解析工具

解析工具通过命令行调用，调用命令如下：

sh ./bin/run.sh read \
-f ./data/0_OriginalPackage/<DataArts调度信息包>.zip \
-o ./data/1_ReaderOutput/<源端探查导出包>.zip \
-t dataartsstudio-reader

其中-c为配置文件路径，-f为输入包路径，-o为源端探查结果包的生成路径，-t为探查插件名称。

例如，当前需要解析项目A：

sh ./bin/run.sh read \
-f ./data/0_OriginalPackage/projectA_DataArtsPkg.zip \
-o ./data/1_ReaderOutput/projectA_ReaderOutput.zip \
-t dataartsstudio-reader

探查工具运行中将打印过程信息，请关注运行过程中是否有报错。

4 查看导出结果

打开./data/1_ReaderOutput/下的生成包ReaderOutput.zip，可预览解析结果。

其中，统计报表是对DataArts导出包中任务流、节点、资源、函数、数据源基本信息的汇总展示。

而data/project文件夹下是对DataArts调度信息数据结构标准化后的结果。

统计报表提供了两项特殊能力：

1、报表中工作流、节点的部分属性被允许更改，允许更改的字段以蓝色字体标识。在下一阶段调度转换中，在初始化阶段，工具将获取表格中的属性变更并使其生效。

2、报表允许通过删除工作流子表中的行，使得在转换时跳过这些工作流（工作流黑名单）。注意！若工作流存在相互依赖关系，相关联的工作流需要同批次转换，不可通过黑名单进行分割。分割会产生异常！

详见：使用调度迁移中的统计报表补充修改调度属性

二、DataArts->DataWorks任务流转换

1 前置条件

探查工具运行完成，WeData调度信息被成功导出，ReaderOutput.zip被成功生成。

（可选，推荐）打开探查导出包，查看统计报表，核对待迁移范围是否被导出完全。

2 转换配置项

2.1 转换配置项模板

• 使用前请删除json中的注释。

{
  "name": "dataartsarts-dw-conveter",
  "self": {
    "if.use.migrationx.before": false,
    "if.use.default.convert": false,
    "if.use.dataworks.newidea": true,
    "param.map": {
      "TODAY": "$[yyyy-mm-dd]",
      "YYYY_MM_DD_1HOUR": "$[yyyy-mm-dd-hh24-1-1/24]",
      "DT": "${yyyy-mm-dd}",
      "DT_HH": "$[yyyy-mm-dd-hh24-1-1/24]",
      "DT_PRE_DAY": "$[yyyy-mm-dd-7]",
      "DT_PRE_BY_DAYS": "",
      "YESTERDAY": "${yyyy-mm-dd}",
      "DT_PRE_2DAYS": "$[yyyy-mm-dd-2]",
      "DT_PRE_3DAYS": "$[yyyy-mm-dd-3]",
      "DT_PRE_4DAYS": "$[yyyy-mm-dd-4]",
      "DT_PRE_5DAYS": "$[yyyy-mm-dd-5]",
      "YYYY_MM_DD_HOUR": "$[yyyy-mm-dd-hh24]"
    },
    "param.value.map": {
      "#{Job.getYesterday(\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-1),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-8),\"yyyyMMddHH\")}": "$[yyyymmddhh24-8]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-1),\"yyyyMMdd\")}": "$[yyyymmdd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-100),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-100]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-185),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-185]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-2),\"yyyyMMdd\")}": "$[yyyymmdd-2]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-31),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-31]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyyMMddHH\")}": "$[yyyymmddhh24]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyyMMdd\")}": "$[yyyymmdd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-0),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-62),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-62]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-2),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-2]"
    },
    "di.datasource.map": [
      {
        "dataArtsLinkDataSourceType": "DWS",
        "dataArtsLinkName": "dws_source1",
        "dwDataSourceType": "HOLOGRES",
        "dwConnectionName": "hologres_source1"
      },
      {
        "dataArtsLinkDataSourceType": "MYSQL",
        "dataArtsLinkName": "mysql_source1",
        "dwDataSourceType": "MYSQL",
        "dwConnectionName": "mysql_source1"
      },
      {
        "dataArtsLinkDataSourceType": "HIVE",
        "dataArtsLinkName": "hive_source1.db1",
        "dwDataSourceType": "ODPS",
        "dwConnectionName": "odps_source1_1"
      }
    ],
    "table.meta.list": [
      {
        "tableName": "",
        "partition": ""
      }
    ],
    "hBasePluginVersion": ""
  },
  "schedule_datasource": {},
  "target_schedule_datasource": {}
}

2.2 调度参数转换配置项

DataArts提供了大量的表达式可在节点的调度参数中使用，如时间表达式。DataArts表达式与DataWorks表达式在语法上有所差异，在调度转换中需要进行处理。参考: 表达式概述_数据治理中心 DataArts Studio_华为云和日期和时间模式_数据治理中心 DataArts Studio_华为云

调度转换配置项中提供了两种替换模式。

1、param.map：指定变量名，将指定变量的值做替换。如配置"TODAY": "$[yyyy-mm-dd]"，则转换工具会将所有名为TODAY的调度变量的值修改为$[yyyy-mm-dd]。

2、param.value.map：指定变量值，将原变量值替换为新变量值。如配置"#{Job.getYesterday(\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-1]"，则转换工具会筛选所有值为#{Job.getYesterday("yyyy-MM-dd")}的变量，并将其值修改为$[yyyy-mm-dd-1]。

本示例中提供了一些常用的映射规则，可供参考。

{
  "self": {
    "param.map": {
      "TODAY": "$[yyyy-mm-dd]",
      "YYYY_MM_DD_1HOUR": "$[yyyy-mm-dd-hh24-1-1/24]",
      "DT": "${yyyy-mm-dd}",
      "DT_HH": "$[yyyy-mm-dd-hh24-1-1/24]",
      "DT_PRE_DAY": "$[yyyy-mm-dd-7]",
      "DT_PRE_BY_DAYS": "",
      "YESTERDAY": "${yyyy-mm-dd}",
      "DT_PRE_2DAYS": "$[yyyy-mm-dd-2]",
      "DT_PRE_3DAYS": "$[yyyy-mm-dd-3]",
      "DT_PRE_4DAYS": "$[yyyy-mm-dd-4]",
      "DT_PRE_5DAYS": "$[yyyy-mm-dd-5]",
      "YYYY_MM_DD_HOUR": "$[yyyy-mm-dd-hh24]"
    },
    "param.value.map": {
      "#{Job.getYesterday(\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-1),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-8),\"yyyyMMddHH\")}": "$[yyyymmddhh24-8]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-1),\"yyyyMMdd\")}": "$[yyyymmdd-1]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-100),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-100]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-185),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-185]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-2),\"yyyyMMdd\")}": "$[yyyymmdd-2]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-31),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-31]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyyMMddHH\")}": "$[yyyymmddhh24]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,0),\"yyyyMMdd\")}": "$[yyyymmdd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-0),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-62),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-62]",
      "#{DateUtil.format(DateUtil.addDays(Job.planTime,-2),\"yyyy-MM-dd\")}": "$[yyyy-mm-dd-2]"
    }
  }

2.3 数据集成转换相关配置项

数据集成配置项主要指定了迁移前后数据源的映射规则，如DataArts Doris数据源迁移后更改为DataWorks Hologres数据源、DataArts Hive数据源迁移后更改为DataWorks Odps（MaxCompute）数据源。此外还补充了若干信息。配置项具体如下：

1. di.datasource.map：迁移前后，DataArts、DataWorks数据源的映射表，dataArtsLinkName、dataArtsLinkDataSourceType分别为DataArts上数据源的名称与类型，dwConnectionName、dwDataSourceType分别为DataWorks上数据源的名称与类型。

   其中Hive数据源较为特殊，在Hive迁移至MaxCompute场景中，Hive数据源每个db通常对应MaxCompute一个空间，因此Hive数据源在dataArtsLinkName中需要附加db信息，如下方示例。

2. table.meta.list：部分数据源迁移后，针对某个表的操作需额外补充分区信息。当前此参数仅在写Hologres数据源时生效。

3. hBasePluginVersion：Hbase版本。参数取值可参考此链接。HBase数据源

{
  "self": {
    "di.datasource.map": [
      {
        "dataArtsLinkDataSourceType": "DWS",
        "dataArtsLinkName": "dws_source1",
        "dwDataSourceType": "HOLOGRES",
        "dwConnectionName": "hologres_source1"
      },
      {
        "dataArtsLinkDataSourceType": "MYSQL",
        "dataArtsLinkName": "mysql_source1",
        "dwDataSourceType": "MYSQL",
        "dwConnectionName": "mysql_source1"
      },
      {
        "dataArtsLinkDataSourceType": "HIVE",
        "dataArtsLinkName": "hive_source1.db1",
        "dwDataSourceType": "ODPS",
        "dwConnectionName": "odps_source1_1"
      }
    ],
    "table.meta.list": [
      {
        "tableName": "",
        "partition": ""
      }
    ],
    "hBasePluginVersion": ""
  }
}

2.4 节点转换规则

工具当前支持转换的DataArts节点包括以下类型：

· CDMJob, HiveSQL, DWSSQL, DLISQL, RDSSQL, SparkSQL

· Shell, DLISpark, MRSSpark, DLFSubJob, RESTAPI, Note, Dummy

当前转换规则固定，不可配置，如需定制请联系LHM迁移工具团队：

· CDMJob（数据集成）：默认转换为DataWorks DI（数据集成）任务。

数据集成脚本（数据集成读写插件配置项）根据di.datasource.map转换。当前工具覆盖了部分数据源相关的CDM->DI配置项转换，在dwDataSourceType中填写DataWorks侧数据源的类型，当前已支持的类型包括：ODPS、ELASTICSEARCH、HBASE、HIVE、HOLOGRES、KAFKA、MONGODB、MYSQL、ORACLE、OSS、POSTGRESQL、SQLSERVER。

· HiveSQL：默认转换为ODPS_SQL（MaxCompute）。

· DWSSQL：默认转换为ODPS_SQL（MaxCompute）。

· DLISQL：默认转换为ODPS_SQL（MaxCompute）。

· RDSSQL：默认转换为ODPS_SQL（MaxCompute）。

· SparkSQL：默认转换为ODPS_SQL（MaxCompute）。

· Shell：默认转换为DIDE_SHELL（通用Shell）。

· DLISpark：默认转换为ODPS_SPARK，部分参数映射。

· MRSSpark：默认转换为ODPS_SPARK，部分参数映射。

· DLFSubJob：默认转换为SUB_PROCESS节点，引用关系获取正常。

· RESTAPI：默认转换为DIDE_SHELL，节点中的HTTP请求通过拼写curl命令实现，当前仅支持GET、POST请求。

· Note：默认转换为VIRTUAL（虚拟节点），保留内容信息。

· Dummy：默认转换为VIRTUAL（虚拟节点），保留内容信息。

· 其余暂不支持的节点，转换为VIRTUAL节点。

DataWorks节点类型可参考此枚举类：

https://github.com/aliyun/dataworks-spec/blob/b0f4a4fd769215d5f81c0bbe990addd7498df5f4/spec/src/main/java/com/aliyun/dataworks/common/spec/domain/dw/types/CodeProgramType.java#L180

3 运行调度转换工具

转换工具通过命令行调用，调用命令如下：

sh ./bin/run.sh convert \
-c ./conf/<你的配置文件>.json \
-f ./data/1_ReaderOutput/<源端探查导出包>.zip \
-o ./data/2_ConverterOutput/<转换结果输出包>.zip \
-t dataartsstudio-dw-conveter

其中-c为配置文件路径，-f为ReaderOutput包存储路径，-o为ConverterOutput包存储路径，-t为转换插件名称。

例如，当前需要转换的DataArts项目A：

sh ./bin/run.sh convert \
-c ./conf/projectA_convert.json \
-f ./data/1_ReaderOutput/projectA_ReaderOutput.zip \
-o ./data/2_ConverterOutput/projectA_ConverterOutput.zip \
-t dataartsstudio-dw-conveter

转换工具运行中将打印过程信息，请关注运行过程中是否有报错。转换完成后将在命令行中打印转换成功与失败的统计信息。注意，部分节点的转换失败不会影响整体转换流程，如遇少量节点转换失败，可在迁移至DataWorks后进行手动修改。

4 查看转换结果

打开./data/2_ConverterOutput/下的生成包ConverterOutput.zip，可预览导出结果。

其中，统计报表是对转换结果任务流、节点、资源、函数、数据源基本信息的汇总展示。

而data/project文件夹是转换完成的调度迁移包本体。

统计报表提供了两项特殊能力：

1、报表中工作流、节点的部分属性被允许更改，允许更改的字段以蓝色字体标识。在下一阶段导入DataWorks时，工具将获取表格中的属性变更并使其生效。

2、报表允许通过删除工作流子表中的行，使得在导入DataWorks时跳过这些工作流（工作流黑名单）。注意！若工作流存在相互依赖关系，相关联的工作流需要同批次导入，不可通过黑名单进行分割。分割会产生异常！

详见：使用调度迁移中的统计报表补充修改调度属性

三、导入DataWorks

LHM迁移工具异构转换已将迁移源端的调度元素转换为DataWorks调度格式，工具得以针对不同的迁移场景提供了统一的上传入口，实现任务流导入DataWorks。

导入工具支持多轮刷写，会自动选择创建/更新任务流（OverWrite模式）。

1 前置条件

1.1 转换成功

转换工具运行完成，源端调度信息被成功转换为DataWorks调度信息，ConverterOutput.zip被成功生成。

（可选，推荐）打开转换输出包，查看统计报表，核对待迁移范围是否被转换成功。

1.2 DataWorks侧配置

DataWorks侧需进行以下动作：

1、创建工作空间。

2、创建AK、SK且保证AK、SK对工作空间具有管理员权限。（强烈建议建立与账号有绑定关系的AK、SK，以便在写入遇到问题时进行排查）

3、在工作空间中建立数据源、绑定计算资源、创建资源组。

4、在工作空间中上传文件资源、创建UDF。

1.3 网络连通性检查

验证能否连接DataWorks Endpoint。

服务接入点列表：

服务接入点

ping dataworks.aliyuncs.com

2 导入配置项

在工程目录的conf文件夹下创建导出配置文件（JSON格式），如writer.json。

• 使用前请删除json中的注释。

{
  "schedule_datasource": {
    "name": "YourDataWorks", //给你的DataWorks数据源起个名字！
    "type": "DataWorks",
    "properties": {
      "endpoint": "dataworks.cn-hangzhou.aliyuncs.com", // 服务接入点
      "project_id": "YourProjectId", // 工作空间ID
      "project_name": "YourProject", // 工作空间名称
      "ak": "************", // AK
      "sk": "************", // SK
    },
    "operaterType": "MANUAL"
  },
  "conf": {
    "di.resource.group.identifier": "Serverless_res_group_***_***", // 调度资源组
    "resource.group.identifier": "Serverless_res_group_***_***", // 数据集成资源组
    "dataworks.node.type.xls": "/Software/bwm-client/conf/CodeProgramType.xls", // DataWorks节点类型表的路径
    "qps.limit": 5 // 向DataWorks发送API请求的QPS上限
  }
}

2.1 服务接入点

根据DataWorks所在Region选择服务接入点，参考文档：

服务接入点

2.2 工作空间ID与名称

打开DataWorks控制台，打开工作空间详情页，从右侧基本信息中获取工作空间ID与名称。

2.3 创建AK、SK并授权

在用户页创建AK、SK，要求对目标DataWorks工作空间拥有管理员读写权限。

权限管理包括两处，如果账号是RAM账号，则需先对RAM账号进行DataWorks操作授权。

权限策略页面：https://ram.console.aliyun.com/policies

然后在DataWorks工作空间中，将工作空间权限赋给账号。

注意！AccessKey可设置网络访问限制策略，请务必保证迁移工具所在机器的IP被允许访问。

2.4 资源组

由DataWorks工作空间详情页左侧菜单栏进入资源组页面，绑定资源组，并获取资源组ID。

通用资源组可用于节点调度，也可用于数据集成。配置项中调度资源组resource.group.identifier和数据集成资源组di.resource.group.identifier可以配置为同一通用资源组。

2.5 QPS设置

工具通过调用DataWorks的API进行导入操作。不同DataWorks版本中的读、写OpenAPI分别有相应的QPS限制和每日调用次数限制，详见链接：使用限制。

DataWorks基础版、标准版、专业版建议填写"qps.limit": 5，企业版建议填写"qps.limit": 20。

注意，请尽可能避免多个导入工具同时运行。

2.6 DataWorks节点类型ID设置

在DataWorks中，部分节点类型在不同Region中被分配了不同的TypeId。具体TypeID以DataWorks数据开发实际界面为准。存在此特性的节点类型以数据库节点为主：数据库节点。

如：MySQL节点在杭州Region的NodeTypeId为1000039、在深圳Region的NodeTypeId为1000041。

为适应上述DataWorks不同Region的差异特性，工具提供了一种可配置的方式，允许用户配置工具所使用的节点TypeId表。

表格通过导入工具的配置项引入：

"conf": {
    "dataworks.node.type.xls": "/Software/bwm-client/conf/CodeProgramType.xls" // DataWorks节点类型表的路径
 }

从DataWorks数据开发界面上获取节点类型Id的方法：在界面上新建一个工作流，并在工作流中新建一个节点，在点击保存后查看工作流的Spec。

若节点类型配置错误，在任务流发布时将提示以下错误。

3 运行DataWorks导入工具

转换工具通过命令行调用，调用命令如下：

sh ./bin/run.sh write \
-c ./conf/<你的配置文件>.json \
-f ./data/2_ConverterOutput/<转换结果输出包>.zip \
-o ./data/4_WriterOutput/<导入结果存储包>.zip \
-t dw-newide-writer

其中-c为配置文件路径，-f为ConverterOutput包存储路径，-o为WriterOutput包存储路径，-t为提交插件名称。

例如，当前需要导入DataWorks的项目A：

sh ./bin/run.sh write \
-c ./conf/projectA_write.json \
-f ./data/2_ConverterOutput/projectA_ConverterOutput.zip \
-o ./data/4_WriterOutput/projectA_WriterOutput.zip \
-t dw-newide-writer

导入工具运行中将打印过程信息，请关注运行过程中是否有报错。导入完成后将在命令行中打印导入成功与失败的统计信息。注意，部分节点的导入失败不会影响整体导入流程，如遇少量节点导入失败，可在DataWorks中进行手动修改。

4 查看导入结果

导入完成后，可在DataWorks中查看导入结果。导入过程中亦可查看工作流逐个导入的过程，如发现问题需要终止导入，可运行jps命令找到BwmClientApp，并使用kill -9终止导入。

5 Q&A

5.1 源端持续在进行开发，这些增量与变更如何提交到DataWorks？

迁移工具为OverWrite模式，重新运行导出、转换、导入可实现将源端增量提交到DataWorks的能力。请注意，工具将根据全路径匹配任务流以选择创建任务流/更新任务流。如需进行变更迁移，请勿移动任务流。

5.2 源端持续在进行开发，同时进行DataWorks上任务流改造与治理，增量迁移时是否会覆盖DataWorks上的变更？

是的，迁移工具为OverWrite模式，建议您在完成迁移后再在DataWorks上进行后续改造。或者采用分批迁移的方式，已迁移等任务流再确认不再刷写后开始DataWorks改造，不同批次之间互相不会影响。

5.3 整个包导入耗时太长，能否只导入一部分

可以，可手动裁剪待导入包来实现部分导入：将data/project/workflow文件夹下需要导入的任务流保留、其他任务流删除，重新压缩回压缩包，再运行导入工具。注意，存在相互依赖的任务流需要捆绑导入，否则任务流间的节点血缘将会丢失。