添加转换流

您可以在转换任务中包括流。流设计器允许您使用源、处理器和目标创建转换流，以定义复杂或简单的转换。

转换流和处理器是逻辑表示。这意味着在 ELT 任务中，每个目标只创建一个表，并且所有处理器都被编译为每个目标的单个 SQL 语句。

支持的目标平台

以下平台支持转换流程：

• Snowflake

• Databricks

• Google BigQuery

• Azure Synapse Analytics

• Microsoft Fabric

• Microsoft SQL Server

• Amazon Redshift

  信息注释以下 Amazon Redshift 功能不可用：UUID 字符串生成和编辑距离。

先决条件

在创建转换流之前，必须：

• 使用要在转换流中使用的引入数据填充存储任务，或注册现有数据。有关如何加载和注册数据的更多信息，请参阅录入数据和注册数据平台上已存在的数据。
• 准备并运行用作转换流中的源的存储或转换任务。运行存储任务不是强制性的，但建议能够在流的每一步显示数据预览。

创建转换流

要创建有效的转换流，至少需要一个源数据集和一个具有已定义键的命名目标。

1. 在数据管道中打开转换数据任务。
2. 在转换中，选择要包含在转换流中的源数据集，然后单击添加转换流。

   将显示添加转换流，您可以在其中提供转换的设置。

3. 在名称中输入目标数据集的名称。

   也可以在描述中添加较长的说明。

4. 在具体化中，选择转换后的输出是否应具体化。您可以选择从数据任务设置中继承设置。

   • 如果开启，将创建表格并处理相关的 ELT 处理。

   • 选择关闭将创建即时执行转换的视图。

5. 增量加载允许您通过应用筛选器或其他条件来调整增量数据加载的查询，以减少使用宏处理的数据集。只有当数据具体化为表时，增量加载才可用。

   • 如果增量加载开启

     任务的第一次执行将执行初始加载，将查询的所有结果插入到目标表中。后续执行将利用您为增量处理定义的筛选器或特定条件执行增量加载。在增量加载期间，任务将只把数据作为更新或插入处理，不管理删除。

   • 如果增量加载关闭

     任务的第一次执行将执行初始加载，将查询的所有结果插入到目标表中。后续执行将通过与目标表进行比较来处理查询的所有结果，并处理新的、更改的或删除的记录。

   信息注释如果查询将选择目标中应该存在的所有记录，请将增量加载设置为关闭。将从目标中删除未选择的记录。

6. 当您准备好创建转换流时，单击添加。

   流设计器打开，目标被创建并作为目标组件显示在流中。

   信息注释必须为流目标提供唯一的名称。如果以后要重命名流目标，请选择目标并在数据集名称中输入新名称。
7. 如果没有为目标定义键，请单击键和可为 null 值字段旁边的编辑。配置窗口打开。
   信息注释键不是从源数据集继承的，必须手动定义。可为空的项是从源数据集继承的，可以修改。
8. 在配置键和可为空的项中，选择要定义为主关键字的列中的键，然后选择要定义为可为空的项的一列或多列中的可为空的项。
9. 单击确认保存更改并关闭配置窗口。

如果转换流的状态有效，则可以关闭该流并准备数据。

您可以稍后在目标设置中更改实体化和增量加载的设置。

• 选择目标，然后单击目标配置中设置旁边的编辑。

添加处理器

您可以将处理器添加到流中。

处理器是您可以添加到流中的组件，以便转换传入数据并将转换后的数据返回到流的下一步。

1. 在流设计器中，选择要在其后添加处理器的流组件。
2. 单击流组件中的 ，然后单击添加处理器，然后选择要添加的处理器。您也可以将处理器从左侧面板拖动到画布上。
提示注释通过从处理器面板中选择处理器，可以显示处理器的简短描述。
3. 根据需要配置处理器，然后单击保存以保存更改并更新数据预览。
   信息注释您可以使用 SQL 和数据预览来预览数据示例。有关更多信息，请参阅预览数据。

可用的处理器

聚合处理器

AI 处理器

清理处理器

串联处理器

筛选器处理器

Fork 处理器

哈希处理器

增量筛选器处理器

联接处理器

数学处理器

数字处理器

删除列处理器

选择列处理器

SQL 表达式处理器

字符串处理器

并集处理器

用例：联接、聚合和筛选 Snowflake 上的数据

在这个用例中，Snowflake 客户数据需要使用处理器进行转换。由于客户信息来自两个数据集，因此您希望首先添加一个联接处理器来组合记录。您还需要使用聚合处理器来计算订单的平均价格，使用筛选器处理器来筛选要保存在输出数据集中的客户记录类型。

第一个数据集基于名为 CUSTOMER_ACCOUNT 的 Snowflake 表，其模式如下所示：

第二个数据集基于名为 CUSTOMER_ORDER 的 Snowflake 表，其模式如下所示：

1. 将一个联接处理器从处理器左侧面板拖动到画布上。
2. 将第二个源链接到联接处理器，以便可以组合来自两个数据集的数据。
3. 配置联结处理器以联接客户 ID 键 (CUSTOMER_ID) 上的两个源数据集。
4. 在联接处理器之后拖动聚合处理器。
5. 配置聚合处理器以计算平均客户购买金额 (ORDER_TOTAL_PRICE)，并将其存储在一个新列中，您可以将该列命名为 avg_order_price，同时按客户细分类型对记录进行分组 (LEFT_CUSTOMER_SEGMENT)。
6. 将一个筛选器处理器从处理器左侧面板拖放到画布上。
7. 配置筛选器处理器以筛选业务客户类型 (Business)。
8. 在筛选器处理器上选择 以打开菜单，然后单击添加不匹配的目标将第二个目标添加到流中。

   此目标将包含与筛选条件不匹配的记录，即单个客户类型 (Individual)。

9. 输入新目标数据集的名称，例如 individual_cust。
10. 检查两个目标中的输出预览：

    business_cust 目标显示业务客户类型的平均订单价格，此处为 157.463687151。

    

    individual_cust 目标显示单独客户类型的平均订单价格，此处为 153.576530612。

    
11. 请确保您的转换流具有有效状态，然后关闭它。
12. 在转换窗口中，单击准备以准备数据。

最佳实践

将源数据集或目标数据集添加到流中时，应在目标数据集的配置面板中设置键和可为空的项。