跳到主要内容
版本: 最新版本-3.5

CREATE ROUTINE LOAD

提示

在此 快速入门 中尝试 Routine Load

Routine Load 可以持续消费来自 Apache Kafka® 的消息，并将数据加载到 StarRocks 中。Routine Load 可以从 Kafka 集群消费 CSV、JSON 和 Avro（v3.0.1 起支持）数据，并通过多种安全协议访问 Kafka，包括 plaintext、ssl、sasl_plaintext 和 sasl_ssl。

本文档介绍 CREATE ROUTINE LOAD 语句的语法、参数和示例。

注意

  • 有关 Routine Load 的应用场景、原理和基本操作，请参阅使用 Routine Load 加载数据。
  • 只有拥有 StarRocks 表的 INSERT 权限的用户才能将数据加载到 StarRocks 表中。如果您没有 INSERT 权限，请按照GRANT中提供的说明，将 INSERT 权限授予用于连接 StarRocks 集群的用户。

语法​

CREATE ROUTINE LOAD <database_name>.<job_name> ON <table_name>
[load_properties]
[job_properties]
FROM data_source
[data_source_properties]

参数​

database_name、job_name、table_name​

database_name

可选。StarRocks 数据库的名称。

job_name

必需。Routine Load 作业的名称。一个表可以接收来自多个 Routine Load 作业的数据。我们建议您使用可识别的信息（例如 Kafka topic 名称和大致作业创建时间）设置有意义的 Routine Load 作业名称，以区分多个 Routine Load 作业。Routine Load 作业的名称在同一数据库中必须是唯一的。

table_name

必需。要加载数据的 StarRocks 表的名称。

load_properties​

可选。数据的属性。语法

[COLUMNS TERMINATED BY '<column_separator>'],
[ROWS TERMINATED BY '<row_separator>'],
[COLUMNS (<column1_name>[, <column2_name>, <column_assignment>, ... ])],
[WHERE <expr>],
[PARTITION (<partition1_name>[, <partition2_name>, ...])]
[TEMPORARY PARTITION (<temporary_partition1_name>[, <temporary_partition2_name>, ...])]

COLUMNS TERMINATED BY

CSV 格式数据的列分隔符。默认列分隔符是 \t（制表符）。例如，您可以使用 COLUMNS TERMINATED BY "," 将列分隔符指定为逗号。

注意

  • 确保此处指定的列分隔符与要摄取的数据中的列分隔符相同。
  • 您可以使用 UTF-8 字符串（例如逗号 (,)、制表符或管道 (|)）作为文本分隔符，其长度不超过 50 字节。
  • 空值用 \N 表示。例如，一个数据记录由三列组成，该数据记录在第一列和第三列中保存数据，但在第二列中不保存数据。在这种情况下，您需要在第二列中使用 \N 来表示空值。这意味着该记录必须编译为 a,\N,b 而不是 a,,b。a,,b 表示记录的第二列包含一个空字符串。

ROWS TERMINATED BY

CSV 格式数据的行分隔符。默认行分隔符是 \n。

COLUMNS

源数据中的列与 StarRocks 表中的列之间的映射。有关详细信息，请参阅本主题中的列映射。

  • column_name: 如果源数据的列可以无需任何计算即可映射到 StarRocks 表的列，则只需指定列名。这些列可以称为映射列。
  • column_assignment: 如果源数据的列无法直接映射到 StarRocks 表的列，并且该列的值必须在使用函数计算后才能加载数据，则必须在 expr 中指定计算函数。这些列可以称为派生列。建议将派生列放在映射列之后，因为 StarRocks 首先解析映射列。

WHERE

筛选条件。只有满足筛选条件的数据才能加载到 StarRocks 中。例如，如果您只想摄取 col1 值大于 100 且 col2 值等于 1000 的行，则可以使用 WHERE col1 > 100 and col2 = 1000。

注意

筛选条件中指定的列可以是源列或派生列。

PARTITION

如果 StarRocks 表分布在分区 p0、p1、p2 和 p3 上，并且您只想将数据加载到 StarRocks 中的 p1、p2 和 p3，并筛选出将存储在 p0 中的数据，则可以将 PARTITION(p1, p2, p3) 指定为筛选条件。默认情况下，如果不指定此参数，数据将加载到所有分区中。示例

PARTITION (p1, p2, p3)

TEMPORARY PARTITION

您要将数据加载到的临时分区的名称。您可以指定多个临时分区，必须用逗号 (,) 分隔。

job_properties​

必需。加载作业的属性。语法

PROPERTIES ("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])
属性必需描述
desired_concurrent_number否单个 Routine Load 作业的预期任务并行度。默认值：3。实际任务并行度由多个参数的最小值决定：min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num)。
  • alive_be_number：活跃 BE 节点的数量。
  • partition_number：要消费的分区数量。
  • desired_concurrent_number：单个 Routine Load 作业的预期任务并行度。默认值：3。
  • max_routine_load_task_concurrent_num：Routine Load 作业的默认最大任务并行度，为 5。请参阅FE 动态参数。
最大实际任务并行度由活跃 BE 节点的数量或要消费的分区数量决定。
max_batch_interval否任务的调度间隔，即任务的执行频率。单位：秒。取值范围：5 ~ 60。默认值：10。建议设置大于 10 的值。如果调度间隔短于 10 秒，则由于加载频率过高，会生成过多的 Tablet 版本。
max_batch_rows否此属性仅用于定义错误检测窗口。该窗口是单个 Routine Load 任务消耗的数据行数。值为 10 * max_batch_rows。默认值为 10 * 200000 = 2000000。Routine Load 任务在错误检测窗口中检测错误数据。错误数据是指 StarRocks 无法解析的数据，例如无效的 JSON 格式数据。
max_error_number否错误检测窗口中允许的最大错误数据行数。如果错误数据行数超过此值，加载作业将暂停。您可以执行SHOW ROUTINE LOAD并通过 ErrorLogUrls 查看错误日志。之后，您可以根据错误日志更正 Kafka 中的错误。默认值为 0，这意味着不允许出现错误行。
注意
  • 当存在过多错误数据行时，最后一批任务将在加载作业暂停之前成功。也就是说，合格的数据将被加载，不合格的数据将被过滤。如果您不想过滤过多不合格的数据行，请设置参数 max_filter_ratio。
  • 错误数据行不包括被 WHERE 子句过滤掉的数据行。
  • 此参数与下一个参数 max_filter_ratio 一起控制最大错误数据记录数。当未设置 max_filter_ratio 时，此参数的值生效。设置 max_filter_ratio 后，一旦错误数据记录数达到此参数设置的阈值或 max_filter_ratio 参数，加载作业将立即暂停。
max_filter_ratio否加载作业的最大错误容忍度。错误容忍度是可以由于数据质量不足而被过滤掉的最大数据记录百分比，占加载作业请求的所有数据记录的百分比。有效值：0 到 1。默认值：1（这意味着它实际上不会生效）。
建议将其设置为 0。这样，如果检测到不合格的数据记录，加载作业将暂停，从而确保数据的正确性。
如果您想忽略不合格的数据记录，可以将此参数设置为大于 0 的值。这样，即使数据文件包含不合格的数据记录，加载作业也可以成功。
注意
  • 当错误数据行数大于 max_filter_ratio 时，最后一批任务将失败。这与 max_error_number 的效果有些不同。
  • 不合格的数据记录不包括被 WHERE 子句过滤掉的数据记录。
  • 此参数与上一个参数 max_error_number 一起控制最大错误数据记录数。当未设置此参数时（其作用与设置 max_filter_ratio = 1 相同），max_error_number 参数的值生效。设置此参数后，一旦错误数据记录数达到此参数设置的阈值或 max_error_number 参数，加载作业将立即暂停。
strict_mode否指定是否启用严格模式。有效值：true 和 false。默认值：false。启用严格模式后，如果加载的数据中列的值为 NULL，但目标表不允许此列的 NULL 值，则该数据行将被过滤掉。
log_rejected_record_num否指定可以记录的不合格数据行的最大数量。此参数从 v3.1 开始支持。有效值：0、-1 和任何非零正整数。默认值：0。
  • 值 0 指定将不记录被过滤掉的数据行。
  • 值 -1 指定将记录所有被过滤掉的数据行。
  • 一个非零正整数（例如 n）指定每个 BE 上最多可以记录 n 个被过滤掉的数据行。
timezone否加载作业使用的时区。默认值：Asia/Shanghai。此参数的值影响 strftime()、alignment_timestamp() 和 from_unixtime() 等函数返回的结果。此参数指定的时区是会话级时区。有关详细信息，请参阅配置时区。
partial_update否是否使用部分更新。有效值：TRUE 和 FALSE。默认值：FALSE，表示禁用此功能。
merge_condition否指定要用作确定是否更新数据的条件的列的名称。仅当要加载到此列中的数据的值大于或等于此列的当前值时，才会更新数据。注意
只有主键表支持条件更新。您指定的列不能是主键列。
format否要加载的数据的格式。有效值：CSV、JSON 和 Avro（v3.0.1 起支持）。默认值：CSV。
trim_space否指定当数据文件为 CSV 格式时是否从数据文件中删除列分隔符之前和之后的空格。类型：BOOLEAN。默认值：false。
对于某些数据库，当您将数据导出为 CSV 格式的数据文件时，空格会添加到列分隔符中。根据空格的位置，这些空格称为前导空格或尾随空格。通过设置 trim_space 参数，您可以使 StarRocks 在数据加载期间删除此类不必要的空格。
请注意，StarRocks 不会删除用一对 enclose 指定的字符包装的字段中的空格（包括前导空格和尾随空格）。例如，以下字段值使用管道符 (|) 作为列分隔符，双引号 (") 作为 enclose 指定的字符：| "Love StarRocks" |。如果您将 trim_space 设置为 true，StarRocks 会将上述字段值处理为 |"Love StarRocks"|。
enclose否指定用于根据RFC4180在数据文件中包装字段值的字符，当数据文件为 CSV 格式时。类型：单字节字符。默认值：NONE。最常用的字符是单引号 (') 和双引号 (")。
由 enclose 指定的字符包装的所有特殊字符（包括行分隔符和列分隔符）都被视为普通符号。StarRocks 可以做更多的事情，因为它允许您将任何单字节字符指定为 enclose 指定的字符。
如果字段值包含 enclose 指定的字符，您可以使用相同的字符来转义该 enclose 指定的字符。例如，您将 enclose 设置为 "，字段值为 a "quoted" c。在这种情况下，您可以将字段值作为 "a ""quoted"" c" 输入到数据文件中。
escape否指定用于转义各种特殊字符（例如行分隔符、列分隔符、转义字符和 enclose 指定的字符）的字符，然后 StarRocks 会将这些字符视为普通字符并将其解析为它们所在的字段值的一部分。类型：单字节字符。默认值：NONE。最常用的字符是斜杠 (\)，在 SQL 语句中必须写成双斜杠 (\\)。
注意
由 escape 指定的字符适用于每对 enclose 指定的字符的内部和外部。
以下是两个示例
  • 当您将 enclose 设置为 " 并将 escape 设置为 \ 时，StarRocks 会将 "say \"Hello world\"" 解析为 say "Hello world"。
  • 假设列分隔符是逗号 (,)。当您将 escape 设置为 \ 时，StarRocks 会将 a, b\, c 解析为两个单独的字段值：a 和 b, c。
strip_outer_array否指定是否剥离 JSON 格式数据的最外层数组结构。有效值：true 和 false。默认值：false。在实际业务场景中，JSON 格式的数据可能具有最外层数组结构，如方括号 [] 所示。在这种情况下，我们建议您将此参数设置为 true，以便 StarRocks 删除最外层方括号 [] 并将每个内部数组加载为单独的数据记录。如果您将此参数设置为 false，StarRocks 会将整个 JSON 格式的数据解析为一个数组并将该数组作为单个数据记录加载。以 JSON 格式的数据 [{"category" : 1, "author" : 2}, {"category" : 3, "author" : 4} ] 为例。如果您将此参数设置为 true，则 {"category" : 1, "author" : 2} 和 {"category" : 3, "author" : 4} 将被解析为两个单独的数据记录并加载到两个 StarRocks 数据行中。
jsonpaths否您要从 JSON 格式的数据加载的字段的名称。此参数的值是有效的 JsonPath 表达式。有关详细信息，请参阅本主题中的StarRocks 表包含其值使用表达式生成的派生列)。
json_root否要加载的 JSON 格式数据的根元素。StarRocks 通过 json_root 提取根节点的元素进行解析。默认情况下，此参数的值为空，表示将加载所有 JSON 格式的数据。有关详细信息，请参阅本主题中的指定要加载的 JSON 格式数据的根元素。
task_consume_second否指定的 Routine Load 作业中每个 Routine Load 任务消耗数据的最长时间。单位：秒。与 FE 动态参数 routine_load_task_consume_second（适用于集群中的所有 Routine Load 作业）不同，此参数特定于单个 Routine Load 作业，这更灵活。此参数自 v3.1.0 起支持。
  • 当未配置 task_consume_second 和 task_timeout_second 时，StarRocks 使用 FE 动态参数 routine_load_task_consume_second 和 routine_load_task_timeout_second 来控制加载行为。
  • 仅配置 task_consume_second 时，task_timeout_second 的默认值计算为 task_consume_second * 4。
  • 仅配置 task_timeout_second 时，task_consume_second 的默认值计算为 task_timeout_second/4。
task_timeout_second否指定的 Routine Load 作业中每个 Routine Load 任务的超时时长。单位：秒。与 FE 动态参数 routine_load_task_timeout_second（适用于集群中的所有 Routine Load 作业）不同，此参数特定于单个 Routine Load 作业，这更灵活。此参数自 v3.1.0 起支持。
  • 当未配置 task_consume_second 和 task_timeout_second 时，StarRocks 使用 FE 动态参数 routine_load_task_consume_second 和 routine_load_task_timeout_second 来控制加载行为。
  • 仅配置 task_timeout_second 时，task_consume_second 的默认值计算为 task_timeout_second/4。
  • 仅配置 task_consume_second 时，task_timeout_second 的默认值计算为 task_consume_second * 4。
pause_on_fatal_parse_error否指定在遇到无法恢复的数据解析错误时是否自动暂停作业。有效值：true 和 false。默认值：false。此参数自 v3.3.12/v3.4.2 起支持。
此类解析错误通常是由非法数据格式引起的，例如
  • 导入 JSON 数组而不设置 strip_outer_array。
  • 导入 JSON 数据，但 Kafka 消息包含非法 JSON，例如 abcd。

data_source、data_source_properties​

必需。数据源和相关属性。

FROM <data_source>
 ("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])

data_source

必需。要加载的数据的源。有效值：KAFKA。

data_source_properties

数据源的属性。

属性必需描述
kafka_broker_list是Kafka 的 broker 连接信息。格式为 <kafka_broker_ip>:<broker_ port>。多个 broker 用逗号 (,) 分隔。Kafka broker 使用的默认端口是 9092。示例："kafka_broker_list" = ""xxx.xx.xxx.xx:9092,xxx.xx.xxx.xx:9092"。
kafka_topic是要消费的 Kafka topic。一个 Routine Load 作业只能消费来自一个 topic 的消息。
kafka_partitions否要消费的 Kafka 分区，例如 "kafka_partitions" = "0, 1, 2, 3"。如果未指定此属性，则默认情况下将消费所有分区。
kafka_offsets否从 Kafka 分区中消费数据的起始 offset，如 kafka_partitions 中指定。如果未指定此属性，Routine Load 作业将从 kafka_partitions 中的最新 offset 开始消费数据。有效值
  • 特定 offset：从特定 offset 开始消费数据。
  • OFFSET_BEGINNING：从最早的可能 offset 开始消费数据。
  • OFFSET_END：从最新的 offset 开始消费数据。
多个起始 offset 用逗号 (,) 分隔，例如 "kafka_offsets" = "1000, OFFSET_BEGINNING, OFFSET_END, 2000"。
property.kafka_default_offsets否所有消费者分区的默认起始 offset。此属性支持的值与 kafka_offsets 属性相同。
confluent.schema.registry.url否Avro schema 注册的 Schema Registry 的 URL。StarRocks 使用此 URL 检索 Avro schema。格式如下
confluent.schema.registry.url = http[s]://[<schema-registry-api-key>:<schema-registry-api-secret>@]<hostname or ip address>[:<port>]

您可以指定其他数据源（Kafka）相关属性，这些属性等效于使用 Kafka 命令行 --property。有关更多支持的属性，请参阅 librdkafka 配置属性中 Kafka 消费者客户端的属性。

注意

如果属性的值是文件名，请在文件名前面添加关键字 FILE:。有关如何创建文件的信息，请参阅CREATE FILE。

  • 指定要消费的所有分区的默认初始 offset
"property.kafka_default_offsets" = "OFFSET_BEGINNING"
  • 指定 Routine Load 作业使用的消费者组的 ID
"property.group.id" = "group_id_0"

如果未指定 property.group.id，StarRocks 会根据 Routine Load 作业的名称生成一个随机值，格式为 {job_name}_{random uuid}，例如 simple_job_0a64fe25-3983-44b2-a4d8-f52d3af4c3e8。

  • 指定 BE 用于访问 Kafka 的安全协议和相关参数

    安全协议可以指定为 plaintext（默认）、ssl、sasl_plaintext 或 sasl_ssl。您需要根据指定的安全协议配置相关参数。

    当安全协议设置为 sasl_plaintext 或 sasl_ssl 时，支持以下 SASL 身份验证机制

    • PLAIN
    • SCRAM-SHA-256 和 SCRAM-SHA-512
    • OAUTHBEARER
    • GSSAPI (Kerberos)

    示例

    • 使用 SSL 安全协议访问 Kafka

      "property.security.protocol" = "ssl", -- Specify the security protocol as SSL.
      "property.ssl.ca.location" = "FILE:ca-cert", -- File or directory path to CA certificate(s) for verifying the kafka broker's key.
      -- If the Kafka server enables client authentication, the following three parameters are also required:
      "property.ssl.certificate.location" = "FILE:client.pem", -- Path to the client's public key used for authentication.
      "property.ssl.key.location" = "FILE:client.key", -- Path to the client's private key used for authentication.
      "property.ssl.key.password" = "xxxxxx" -- Password for the client's private key.

    • 使用 SASL_PLAINTEXT 安全协议和 SASL/PLAIN 身份验证机制访问 Kafka

      "property.security.protocol" = "SASL_PLAINTEXT", -- Specify the security protocol as SASL_PLAINTEXT.
      "property.sasl.mechanism" = "PLAIN", -- Specify the SASL mechanism as PLAIN which is a simple username/password authentication mechanism.
      "property.sasl.username" = "admin",  -- SASL username.
      "property.sasl.password" = "xxxxxx"  -- SASL password.

    • 使用 SASL_PLAINTEXT 安全协议和 SASL/GSSAPI (Kerberos) 身份验证机制访问 Kafka

      "property.security.protocol" = "SASL_PLAINTEXT", -- Specify the security protocol as SASL_PLAINTEXT.
      "property.sasl.mechanism" = "GSSAPI", -- Specify the SASL authentication mechanism as GSSAPI. Default value is GSSAPI.
      "property.sasl.kerberos.service.name" = "kafka", -- The broker service name. Default value is kafka.
      "property.sasl.kerberos.keytab" = "/home/starrocks/starrocks.keytab", -- The client keytab location.
      "property.sasl.kerberos.principal" = "starrocks@YOUR.COM" -- The Kerberos principal.
      注意

      • 自 StarRocks v3.1.4 起，支持 SASL/GSSAPI (Kerberos) 身份验证。

      • 需要在 BE 机器上安装 SASL 相关模块。

        # Debian/Ubuntu:
        sudo apt-get install libsasl2-modules-gssapi-mit libsasl2-dev
        # CentOS/Redhat:
        sudo yum install cyrus-sasl-gssapi cyrus-sasl-devel

FE 和 BE 配置项​

有关与 Routine Load 相关的 FE 和 BE 配置项，请参阅配置项。

列映射​

配置用于加载 CSV 格式数据的列映射​

如果 CSV 格式数据的列可以按顺序一一映射到 StarRocks 表的列，则无需配置数据和 StarRocks 表之间的列映射。

如果 CSV 格式数据的列无法按顺序一一映射到 StarRocks 表的列，则需要使用 columns 参数配置数据文件和 StarRocks 表之间的列映射。这包括以下两种用例

  • 列数相同但列顺序不同。此外，数据文件中的数据在加载到匹配的 StarRocks 表列之前不需要通过函数计算。

    • 在 columns 参数中，您需要以数据文件列的排列顺序指定 StarRocks 表列的名称。

    • 例如，StarRocks 表由三列组成，按顺序为 col1、col2 和 col3，数据文件也由三列组成，可以按顺序映射到 StarRocks 表列 col3、col2 和 col1。在这种情况下，您需要指定 "columns: col3, col2, col1"。

  • 列数不同且列顺序不同。此外，数据文件中的数据在加载到匹配的 StarRocks 表列之前需要通过函数计算。

    在 columns 参数中，您需要以数据文件列的排列顺序指定 StarRocks 表列的名称，并指定要用于计算数据的函数。以下是两个示例

    • StarRocks 表由三列组成，按顺序为 col1、col2 和 col3。数据文件由四列组成，其中前三列可以按顺序映射到 StarRocks 表列 col1、col2 和 col3，第四列无法映射到任何 StarRocks 表列。在这种情况下，您需要为数据文件的第四列临时指定一个名称，并且临时名称必须与任何 StarRocks 表列名不同。例如，您可以指定 "columns: col1, col2, col3, temp"，其中数据文件的第四列临时命名为 temp。
    • StarRocks 表由三列组成，按顺序为 year、month 和 day。数据文件仅由一列组成，该列容纳 yyyy-mm-dd hh:mm:ss 格式的日期和时间值。在这种情况下，您可以指定 "columns: col, year = year(col), month=month(col), day=day(col)"，其中 col 是数据文件列的临时名称，函数 year = year(col)、month=month(col) 和 day=day(col) 用于从数据文件列 col 中提取数据，并将数据加载到映射的 StarRocks 表列中。例如，year = year(col) 用于从数据文件列 col 中提取 yyyy 数据并将数据加载到 StarRocks 表列 year 中。

有关更多示例，请参阅配置列映射。

配置用于加载 JSON 格式或 Avro 格式数据的列映射​

注意

自 v3.0.1 起，StarRocks 支持使用 Routine Load 加载 Avro 数据。当您加载 JSON 或 Avro 数据时，列映射和转换的配置是相同的。因此，在本节中，JSON 数据用作示例来介绍配置。

如果 JSON 格式数据的键与 StarRocks 表的列具有相同的名称，则可以使用简单模式加载 JSON 格式的数据。在简单模式下，您不需要指定 jsonpaths 参数。此模式要求 JSON 格式的数据必须是对象，如花括号 {} 所示，例如 {"category": 1, "author": 2, "price": "3"}。在此示例中，category、author 和 price 是键名，这些键可以按名称一一映射到 StarRocks 表的列 category、author 和 price。有关示例，请参阅简单模式。

如果 JSON 格式数据的键与 StarRocks 表的列的名称不同，则可以使用匹配模式加载 JSON 格式的数据。在匹配模式下，您需要使用 jsonpaths 和 COLUMNS 参数来指定 JSON 格式数据和 StarRocks 表之间的列映射

  • 在 jsonpaths 参数中，按照 JSON 格式数据中排列的顺序指定 JSON 键。
  • 在 COLUMNS 参数中，指定 JSON 键和 StarRocks 表列之间的映射
    • COLUMNS 参数中指定的列名按顺序一一映射到 JSON 格式的数据。
    • COLUMNS 参数中指定的列名按名称一一映射到 StarRocks 表列。

有关示例，请参阅StarRocks 表包含其值使用表达式生成的派生列。

示例​

加载 CSV 格式的数据​

本节以 CSV 格式的数据为例，介绍如何使用各种参数设置和组合来满足您的各种加载需求。

准备数据集

假设您要从名为 ordertest1 的 Kafka topic 加载 CSV 格式的数据。数据集中的每条消息都包含六列：订单 ID、付款日期、客户姓名、国籍、性别和价格。

2020050802,2020-05-08,Johann Georg Faust,Deutschland,male,895
2020050802,2020-05-08,Julien Sorel,France,male,893
2020050803,2020-05-08,Dorian Grey,UK,male,1262
2020050901,2020-05-09,Anna Karenina,Russia,female,175
2020051001,2020-05-10,Tess Durbeyfield,US,female,986
2020051101,2020-05-11,Edogawa Conan,japan,male,8924

创建表

根据 CSV 格式数据的列，在数据库 example_db 中创建名为 example_tbl1 的表。

CREATE TABLE example_db.example_tbl1 ( 
    `order_id` bigint NOT NULL COMMENT "Order ID",
    `pay_dt` date NOT NULL COMMENT "Payment date", 
    `customer_name` varchar(26) NULL COMMENT "Customer name", 
    `nationality` varchar(26) NULL COMMENT "Nationality", 
    `gender` varchar(26) NULL COMMENT "Gender", 
    `price` double NULL COMMENT "Price") 
DUPLICATE KEY (order_id,pay_dt) 
DISTRIBUTED BY HASH(`order_id`);

从指定分区的指定偏移量开始消费数据​

如果 Routine Load 作业需要从指定分区和偏移量开始消费数据，则需要配置参数 kafka_partitions 和 kafka_offsets。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1",
    "kafka_partitions" ="0,1,2,3,4", -- partitions to be consumed
    "kafka_offsets" = "1000, OFFSET_BEGINNING, OFFSET_END, 2000" -- corresponding initial offsets
);

通过增加任务并行度来提高加载性能​

为了提高加载性能并避免累积消费，您可以通过在创建 Routine Load 作业时增加 desired_concurrent_number 值来增加任务并行度。任务并行度允许将一个 Routine Load 作业拆分为尽可能多的并行任务。

请注意，实际的任务并行度由以下多个参数中的最小值决定

min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num)

注意

最大实际任务并行度是存活 BE 节点数或要消费的分区数。

因此，当存活 BE 节点数和要消费的分区数大于其他两个参数 max_routine_load_task_concurrent_num 和 desired_concurrent_number 的值时，您可以增加其他两个参数的值以增加实际的任务并行度。

假设要消费的分区数为 7，存活 BE 节点数为 5，并且 max_routine_load_task_concurrent_num 是默认值 5。 如果要增加实际的任务并行度，可以将 desired_concurrent_number 设置为 5 （默认值为 3）。 在这种情况下，实际的任务并行度 min(5,7,5,5) 被配置为 5。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
PROPERTIES
(
"desired_concurrent_number" = "5" -- set the value of desired_concurrent_number to 5
)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1"
);

配置列映射​

如果 CSV 格式数据中的列顺序与目标表中的列不一致，假设 CSV 格式数据中的第五列不需要导入到目标表，则需要通过 COLUMNS 参数指定 CSV 格式数据和目标表之间的列映射。

目标数据库和表

根据 CSV 格式数据中的列在目标数据库 example_db 中创建目标表 example_tbl2。 在此方案中，您需要创建五个列，与 CSV 格式数据中的五个列相对应，除了存储性别的第五列。

CREATE TABLE example_db.example_tbl2 ( 
    `order_id` bigint NOT NULL COMMENT "Order ID",
    `pay_dt` date NOT NULL COMMENT "Payment date", 
    `customer_name` varchar(26) NULL COMMENT "Customer name", 
    `nationality` varchar(26) NULL COMMENT "Nationality", 
    `price` double NULL COMMENT "Price"
) 
DUPLICATE KEY (order_id,pay_dt) 
DISTRIBUTED BY HASH(order_id);

Routine Load 作业

在此示例中，由于 CSV 格式数据中的第五列不需要加载到目标表，因此第五列在 COLUMNS 中暂时命名为 temp_gender，其他列直接映射到表 example_tbl2。

CREATE ROUTINE LOAD example_db.example_tbl2_ordertest1 ON example_tbl2
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1"
);

设置过滤条件​

如果要仅加载符合特定条件的数据，可以在 WHERE 子句中设置过滤条件，例如 price > 100.

CREATE ROUTINE LOAD example_db.example_tbl2_ordertest1 ON example_tbl2
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price),
WHERE price > 100 -- set the filter condition
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1"
);

启用严格模式以过滤掉具有 NULL 值的行​

在 PROPERTIES 中，您可以设置 "strict_mode" = "true"，这意味着 Routine Load 作业处于严格模式。 如果源列中存在 NULL 值，但目标 StarRocks 表列不允许 NULL 值，则包含源列中 NULL 值的行将被过滤掉。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
PROPERTIES
(
"strict_mode" = "true" -- enable the strict mode
)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1"
);

设置错误容忍度​

如果您的业务场景对不合格数据的容忍度较低，则需要通过配置参数 max_batch_rows 和 max_error_number 来设置错误检测窗口和最大错误数据行数。 当错误检测窗口内的错误数据行数超过 max_error_number 的值时，Routine Load 作业将暂停。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
PROPERTIES
(
"max_batch_rows" = "100000",-- The value of max_batch_rows multiplied by 10 equals the error detection window.
"max_error_number" = "100" -- The maximum number of error data rows allowed within an error detection window.
)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1"
);

如果需要指定 BE 用于访问 Kafka 的安全协议为 SSL，则需要配置 "property.security.protocol" = "ssl" 和相关参数。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
PROPERTIES
(
    "format" = "json"
)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1",
    -- Specify the security protocol as SSL.
    "property.security.protocol" = "ssl",
    -- The location of the CA certificate.
    "property.ssl.ca.location" = "FILE:ca-cert",
    -- If authentication is enabled for Kafka clients, you need to configure the following properties:
    -- The location of the Kafka client's public key.
    "property.ssl.certificate.location" = "FILE:client.pem",
    -- The location of the Kafka client's private key.
    "property.ssl.key.location" = "FILE:client.key",
    -- The password to the Kafka client's private key.
    "property.ssl.key.password" = "abcdefg"
);

设置 trim_space、enclose 和 escape​

假设您要从名为 test_csv 的 Kafka 主题加载 CSV 格式的数据。 数据集中的每条消息都包含六列：订单 ID、付款日期、客户名称、国籍、性别和价格。

 "2020050802" , "2020-05-08" , "Johann Georg Faust" , "Deutschland" , "male" , "895"
 "2020050802" , "2020-05-08" , "Julien Sorel" , "France" , "male" , "893"
 "2020050803" , "2020-05-08" , "Dorian Grey\,Lord Henry" , "UK" , "male" , "1262"
 "2020050901" , "2020-05-09" , "Anna Karenina" , "Russia" , "female" , "175"
 "2020051001" , "2020-05-10" , "Tess Durbeyfield" , "US" , "female" , "986"
 "2020051101" , "2020-05-11" , "Edogawa Conan" , "japan" , "male" , "8924"

如果要将 Kafka 主题 test_csv 中的所有数据加载到 example_tbl1 中，目的是删除列分隔符前后面的空格，并将 enclose 设置为 "，escape 设置为 \，请运行以下命令

CREATE ROUTINE LOAD example_db.example_tbl1_test_csv ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, gender, price)
PROPERTIES
(
    "trim_space"="true",
    "enclose"="\"",
    "escape"="\\",
)
FROM KAFKA
(
    "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic"="test_csv",
    "property.kafka_default_offsets"="OFFSET_BEGINNING"
);

加载 JSON 格式的数据​

StarRocks 表列名与 JSON 键名一致​

准备数据集

例如，Kafka 主题 ordertest2 中存在以下 JSON 格式的数据。

{"commodity_id": "1", "customer_name": "Mark Twain", "country": "US","pay_time": 1589191487,"price": 875}
{"commodity_id": "2", "customer_name": "Oscar Wilde", "country": "UK","pay_time": 1589191487,"price": 895}
{"commodity_id": "3", "customer_name": "Antoine de Saint-Exupéry","country": "France","pay_time": 1589191487,"price": 895}

注意 每个 JSON 对象必须在一条 Kafka 消息中。 否则，将发生指示解析 JSON 格式数据失败的错误。

目标数据库和表

在 StarRocks 集群中的目标数据库 example_db 中创建表 example_tbl3。 列名与 JSON 格式数据中的键名一致。

CREATE TABLE example_db.example_tbl3 ( 
    commodity_id varchar(26) NULL, 
    customer_name varchar(26) NULL, 
    country varchar(26) NULL, 
    pay_time bigint(20) NULL, 
    price double SUM NULL COMMENT "Price") 
AGGREGATE KEY(commodity_id,customer_name,country,pay_time)
DISTRIBUTED BY HASH(commodity_id);

Routine Load 作业

您可以使用 Routine Load 作业的简单模式。 也就是说，创建 Routine Load 作业时，您不需要指定 jsonpaths 和 COLUMNS 参数。 StarRocks 根据目标表 example_tbl3 的列名，提取 Kafka 集群主题 ordertest2 中 JSON 格式数据的键，并将 JSON 格式的数据加载到目标表中。

CREATE ROUTINE LOAD example_db.example_tbl3_ordertest2 ON example_tbl3
PROPERTIES
(
    "format" = "json"
 )
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest2"
);

注意

  • 如果 JSON 格式数据的最外层是一个数组结构，则需要在 PROPERTIES 中设置 "strip_outer_array"="true" 来剥离最外层的数组结构。 此外，当您需要指定 jsonpaths 时，由于 JSON 格式数据的最外层数组结构已被剥离，因此整个 JSON 格式数据的根元素是扁平化的 JSON 对象。
  • 您可以使用 json_root 来指定 JSON 格式数据的根元素。

StarRocks 表包含派生列，其值通过使用表达式生成​

准备数据集

例如，Kafka 集群的主题 ordertest2 中存在以下 JSON 格式的数据。

{"commodity_id": "1", "customer_name": "Mark Twain", "country": "US","pay_time": 1589191487,"price": 875}
{"commodity_id": "2", "customer_name": "Oscar Wilde", "country": "UK","pay_time": 1589191487,"price": 895}
{"commodity_id": "3", "customer_name": "Antoine de Saint-Exupéry","country": "France","pay_time": 1589191487,"price": 895}

目标数据库和表

在 StarRocks 集群的数据库 example_db 中创建一个名为 example_tbl4 的表。 列 pay_dt 是一个派生列，其值通过计算 JSON 格式数据中键 pay_time 的值来生成。

CREATE TABLE example_db.example_tbl4 ( 
    `commodity_id` varchar(26) NULL, 
    `customer_name` varchar(26) NULL, 
    `country` varchar(26) NULL,
    `pay_time` bigint(20) NULL,  
    `pay_dt` date NULL, 
    `price` double SUM NULL) 
AGGREGATE KEY(`commodity_id`,`customer_name`,`country`,`pay_time`,`pay_dt`) 
DISTRIBUTED BY HASH(`commodity_id`);

Routine Load 作业

您可以使用 Routine Load 作业的匹配模式。 也就是说，创建 Routine Load 作业时，您需要指定 jsonpaths 和 COLUMNS 参数。

您需要在 jsonpaths 参数中指定 JSON 格式数据的键，并按顺序排列它们。

并且由于 JSON 格式数据中键 pay_time 的值需要在存储在 example_tbl4 表的 pay_dt 列中之前转换为 DATE 类型，因此需要在 COLUMNS 中使用 pay_dt=from_unixtime(pay_time,'%Y%m%d') 指定计算。 JSON 格式数据中的其他键的值可以直接映射到 example_tbl4 表。

CREATE ROUTINE LOAD example_db.example_tbl4_ordertest2 ON example_tbl4
COLUMNS(commodity_id, customer_name, country, pay_time, pay_dt=from_unixtime(pay_time, '%Y%m%d'), price)
PROPERTIES
(
    "format" = "json",
    "jsonpaths" = "[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]"
 )
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest2"
);

注意

  • 如果 JSON 数据的最外层是一个数组结构，则需要在 PROPERTIES 中设置 "strip_outer_array"="true" 来剥离最外层的数组结构。 此外，当您需要指定 jsonpaths 时，由于 JSON 数据的最外层数组结构已被剥离，因此整个 JSON 数据的根元素是扁平化的 JSON 对象。
  • 您可以使用 json_root 来指定 JSON 格式数据的根元素。

StarRocks 表包含派生列，其值通过使用 CASE 表达式生成​

准备数据集

例如，Kafka 主题 topic-expr-test 中存在以下 JSON 格式的数据。

{"key1":1, "key2": 21}
{"key1":12, "key2": 22}
{"key1":13, "key2": 23}
{"key1":14, "key2": 24}

目标数据库和表

在 StarRocks 集群的数据库 example_db 中创建一个名为 tbl_expr_test 的表。 目标表 tbl_expr_test 包含两列，其中 col2 列的值需要通过对 JSON 数据使用 case 表达式来计算。

CREATE TABLE tbl_expr_test (
    col1 string, col2 string)
DISTRIBUTED BY HASH (col1);

Routine Load 作业

由于目标表中的 col2 列中的值是使用 CASE 表达式生成的，因此您需要在 Routine load 作业的 COLUMNS 参数中指定相应的表达式。

CREATE ROUTINE LOAD rl_expr_test ON tbl_expr_test
COLUMNS (
      key1,
      key2,
      col1 = key1,
      col2 = CASE WHEN key1 = "1" THEN "key1=1" 
                  WHEN key1 = "12" THEN "key1=12"
                  ELSE "nothing" END) 
PROPERTIES ("format" = "json")
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "topic-expr-test"
);

查询 StarRocks 表

查询 StarRocks 表。 结果表明 col2 列中的值是 CASE 表达式的输出。

MySQL [example_db]> SELECT * FROM tbl_expr_test;
+------+---------+
| col1 | col2    |
+------+---------+
| 1    | key1=1  |
| 12   | key1=12 |
| 13   | nothing |
| 14   | nothing |
+------+---------+
4 rows in set (0.015 sec)

指定要加载的 JSON 格式数据的根元素​

您需要使用 json_root 来指定要加载的 JSON 格式数据的根元素，并且该值必须是有效的 JsonPath 表达式。

准备数据集

例如，Kafka 集群的主题 ordertest3 中存在以下 JSON 格式的数据。 并且要加载的 JSON 格式数据的根元素是 $.RECORDS。

{"RECORDS":[{"commodity_id": "1", "customer_name": "Mark Twain", "country": "US","pay_time": 1589191487,"price": 875},{"commodity_id": "2", "customer_name": "Oscar Wilde", "country": "UK","pay_time": 1589191487,"price": 895},{"commodity_id": "3", "customer_name": "Antoine de Saint-Exupéry","country": "France","pay_time": 1589191487,"price": 895}]}

目标数据库和表

在 StarRocks 集群的数据库 example_db 中创建一个名为 example_tbl3 的表。

CREATE TABLE example_db.example_tbl3 ( 
    commodity_id varchar(26) NULL, 
    customer_name varchar(26) NULL, 
    country varchar(26) NULL, 
    pay_time bigint(20) NULL, 
    price double SUM NULL) 
AGGREGATE KEY(commodity_id,customer_name,country,pay_time) 
ENGINE=OLAP
DISTRIBUTED BY HASH(commodity_id);

Routine Load 作业

您可以在 PROPERTIES 中设置 "json_root" = "$.RECORDS" 以指定要加载的 JSON 格式数据的根元素。 此外，由于要加载的 JSON 格式数据是数组结构，因此还必须设置 "strip_outer_array" = "true" 以剥离最外层的数组结构。

CREATE ROUTINE LOAD example_db.example_tbl3_ordertest3 ON example_tbl3
PROPERTIES
(
    "format" = "json",
    "json_root" = "$.RECORDS",
    "strip_outer_array" = "true"
 )
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest2"
);

加载 Avro 格式的数据​

自 v3.0.1 起，StarRocks 支持使用 Routine Load 加载 Avro 数据。

Avro schema 很简单​

假设 Avro schema 相对简单，并且您需要加载 Avro 数据的所有字段。

准备数据集

  • Avro schema

    1. 创建以下 Avro schema 文件 avro_schema1.avsc

      {
          "type": "record",
          "name": "sensor_log",
          "fields" : [
              {"name": "id", "type": "long"},
              {"name": "name", "type": "string"},
              {"name": "checked", "type" : "boolean"},
              {"name": "data", "type": "double"},
              {"name": "sensor_type", "type": {"type": "enum", "name": "sensor_type_enum", "symbols" : ["TEMPERATURE", "HUMIDITY", "AIR-PRESSURE"]}}  
          ]
      }

    2. 在 Schema Registry 中注册 Avro schema。

  • Avro 数据

准备 Avro 数据并将其发送到 Kafka 主题 topic_1。

目标数据库和表

根据 Avro 数据的字段，在 StarRocks 集群的目标数据库 sensor 中创建表 sensor_log1。 表的列名必须与 Avro 数据中的字段名匹配。 有关 Avro 数据加载到 StarRocks 中的数据类型映射，请参见 [数据类型映射](#Data types mapping) 。

CREATE TABLE sensor.sensor_log1 ( 
    `id` bigint NOT NULL COMMENT "sensor id",
    `name` varchar(26) NOT NULL COMMENT "sensor name", 
    `checked` boolean NOT NULL COMMENT "checked", 
    `data` double NULL COMMENT "sensor data", 
    `sensor_type` varchar(26) NOT NULL COMMENT "sensor type"
) 
ENGINE=OLAP 
DUPLICATE KEY (id) 
DISTRIBUTED BY HASH(`id`);

Routine Load 作业

您可以使用 Routine Load 作业的简单模式。 也就是说，创建 Routine Load 作业时，您不需要指定参数 jsonpaths。 执行以下语句以提交一个名为 sensor_log_load_job1 的 Routine Load 作业，以消费 Kafka 主题 topic_1 中的 Avro 消息并将数据加载到数据库 sensor 中的表 sensor_log1 中。

CREATE ROUTINE LOAD sensor.sensor_log_load_job1 ON sensor_log1  
PROPERTIES  
(  
  "format" = "avro"  
)  
FROM KAFKA  
(  
  "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>,...",
  "confluent.schema.registry.url" = "http://172.xx.xxx.xxx:8081",  
  "kafka_topic"= "topic_1",  
  "kafka_partitions" = "0,1,2,3,4,5",  
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"  
);

Avro schema 包含嵌套的 record-type 字段​

假设 Avro schema 包含嵌套的 record-type 字段，并且您需要将嵌套的 record-type 字段中的子字段加载到 StarRocks 中。

准备数据集

  • Avro schema

    1. 创建以下 Avro schema 文件 avro_schema2.avsc。 外部 Avro 记录包括五个字段，它们依次是 id、name、checked、sensor_type 和 data。 并且字段 data 具有嵌套的记录 data_record。

      {
          "type": "record",
          "name": "sensor_log",
          "fields" : [
              {"name": "id", "type": "long"},
              {"name": "name", "type": "string"},
              {"name": "checked", "type" : "boolean"},
              {"name": "sensor_type", "type": {"type": "enum", "name": "sensor_type_enum", "symbols" : ["TEMPERATURE", "HUMIDITY", "AIR-PRESSURE"]}},
              {"name": "data", "type": 
                  {
                      "type": "record",
                      "name": "data_record",
                      "fields" : [
                          {"name": "data_x", "type" : "boolean"},
                          {"name": "data_y", "type": "long"}
                      ]
                  }
              }
          ]
      }

    2. 在 Schema Registry 中注册 Avro schema。

  • Avro 数据

准备 Avro 数据并将其发送到 Kafka 主题 topic_2。

目标数据库和表

根据 Avro 数据的字段，在 StarRocks 集群的目标数据库 sensor 中创建表 sensor_log2。

假设除了加载外部 Record 的字段 id、name、checked 和 sensor_type 之外，还需要加载嵌套 Record data_record 中的子字段 data_y。

CREATE TABLE sensor.sensor_log2 ( 
    `id` bigint NOT NULL COMMENT "sensor id",
    `name` varchar(26) NOT NULL COMMENT "sensor name", 
    `checked` boolean NOT NULL COMMENT "checked", 
    `sensor_type` varchar(26) NOT NULL COMMENT "sensor type",
    `data_y` long NULL COMMENT "sensor data" 
) 
ENGINE=OLAP 
DUPLICATE KEY (id) 
DISTRIBUTED BY HASH(`id`);

Routine Load 作业

提交加载作业，使用 jsonpaths 指定需要加载的 Avro 数据的字段。 请注意，对于嵌套 Record 中的子字段 data_y，您需要将其 jsonpath 指定为 "$.data.data_y"。

CREATE ROUTINE LOAD sensor.sensor_log_load_job2 ON sensor_log2  
PROPERTIES  
(  
  "format" = "avro",
  "jsonpaths" = "[\"$.id\",\"$.name\",\"$.checked\",\"$.sensor_type\",\"$.data.data_y\"]"
)  
FROM KAFKA  
(  
  "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>,...",
  "confluent.schema.registry.url" = "http://172.xx.xxx.xxx:8081",  
  "kafka_topic" = "topic_1",  
  "kafka_partitions" = "0,1,2,3,4,5",  
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"  
);

Avro schema 包含 Union 字段​

准备数据集

假设 Avro schema 包含 Union 字段，并且您需要将 Union 字段加载到 StarRocks 中。

  • Avro schema

    1. 创建以下 Avro schema 文件 avro_schema3.avsc。 外部 Avro 记录包括五个字段，它们依次是 id、name、checked、sensor_type 和 data。 并且字段 data 的类型为 Union，包括两个元素，null 和嵌套的记录 data_record。

      {
          "type": "record",
          "name": "sensor_log",
          "fields" : [
              {"name": "id", "type": "long"},
              {"name": "name", "type": "string"},
              {"name": "checked", "type" : "boolean"},
              {"name": "sensor_type", "type": {"type": "enum", "name": "sensor_type_enum", "symbols" : ["TEMPERATURE", "HUMIDITY", "AIR-PRESSURE"]}},
              {"name": "data", "type": [null,
                      {
                          "type": "record",
                          "name": "data_record",
                          "fields" : [
                              {"name": "data_x", "type" : "boolean"},
                              {"name": "data_y", "type": "long"}
                          ]
                      }
                  ]
              }
          ]
      }

    2. 在 Schema Registry 中注册 Avro schema。

  • Avro 数据

准备 Avro 数据并将其发送到 Kafka 主题 topic_3。

目标数据库和表

根据 Avro 数据的字段，在 StarRocks 集群的目标数据库 sensor 中创建表 sensor_log3。

假设除了加载外部 Record 的字段 id、name、checked 和 sensor_type 之外，还需要加载 Union 类型字段 data 中元素 data_record 的字段 data_y。

CREATE TABLE sensor.sensor_log3 ( 
    `id` bigint NOT NULL COMMENT "sensor id",
    `name` varchar(26) NOT NULL COMMENT "sensor name", 
    `checked` boolean NOT NULL COMMENT "checked", 
    `sensor_type` varchar(26) NOT NULL COMMENT "sensor type",
    `data_y` long NULL COMMENT "sensor data" 
) 
ENGINE=OLAP 
DUPLICATE KEY (id) 
DISTRIBUTED BY HASH(`id`);

Routine Load 作业

提交加载作业，使用 jsonpaths 指定需要在 Avro 数据中加载的字段。 请注意，对于字段 data_y，您需要将其 jsonpath 指定为 "$.data.data_y"。

CREATE ROUTINE LOAD sensor.sensor_log_load_job3 ON sensor_log3  
PROPERTIES  
(  
  "format" = "avro",
  "jsonpaths" = "[\"$.id\",\"$.name\",\"$.checked\",\"$.sensor_type\",\"$.data.data_y\"]"
)  
FROM KAFKA  
(  
  "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>,...",
  "confluent.schema.registry.url" = "http://172.xx.xxx.xxx:8081",  
  "kafka_topic" = "topic_1",  
  "kafka_partitions" = "0,1,2,3,4,5",  
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"  
);

当 Union 类型字段 data 的值为 null 时，加载到 StarRocks 表的 data_y 列中的值为 null。 当 Union 类型字段 data 的值为数据记录时，加载到 data_y 列中的值的类型为 Long。