备案控制台
文档
产品文档
输入文档关键字查找
首页 云数据库 SelectDB 版 开发指南 数据操作 数据导入 Routine Load

Routine Load

更新时间：
一键部署
提交缺陷
我的收藏

本文介绍如何通过Routine Load导入数据至云数据库 SelectDB 版实例。

背景信息

Routine Load支持提交一个常驻的导入任务，通过不断地从指定的数据源读取数据，将数据持续地导入到云数据库 SelectDB 版中。

Routine Load功能只支持从Kafka数据源读取数据。首先需要创建一个Routine Load作业，作业会通过例行调度，不断地发送任务，每个任务会消费一定数量的Kafka消息。

注意事项

  • Routine Load支持访问无认证的Kafka，和通过SSL方式认证的Kafka。

  • 支持的消息格式为CSV或JSON格式。CSV的格式，每一个Message为一行，且行尾不包含换行符。

  • 默认支持Kafka 0.10.0.0（含）以上版本。如果使用Kafka 0.10.0.0以下版本（0.9.0，0.8.2，0.8.1，0.8.0），需要修改BE的配置，将kafka_broker_version_fallback的值设置为要兼容的旧版本，或者在创建Routine Load的时候直接设置property.broker.version.fallback的值为要兼容的旧版本，使用旧版本的代价是Routine Load的部分新特性可能无法使用，如根据时间设置Kafka分区的offset。

创建导入

语法

CREATE ROUTINE LOAD [db.]job_name ON tbl_name
[merge_type]
[load_properties]
[job_properties]
FROM data_source [data_source_properties]

参数说明

参数名称

参数说明

[db.]job_name

导入作业的名称。在同一个Database内，相同名称的job只能运行一个。

tbl_name

指定导入的表的名称。

merge_type

指定数据合并类型。默认为APPEND，表示导入的数据都是普通的追加写操作。MERGE和DELETE类型仅适用于Unique Key模型表。其中MERGE类型需要配合[DELETE ON]语句使用，以标注Delete Flag列。而DELETE类型则表示导入的所有数据皆为删除数据。

load_properties

指定导入数据处理相关参数。详细参数说明，请参见load_properties参数说明。

job_properties

指定导入作业相关参数。详细参数说明，请参见job_properties参数说明。

data_source_properties

指定数据源的类型。详细参数说明，请参见data_source_properties参数说明。

load_properties参数说明

[column_separator],
[columns_mapping],
[preceding_filter],
[where_predicates],
[partitions],
[DELETE ON],
[ORDER BY]

参数名称

示例值

参数说明

column_separator

COLUMNS TERMINATED BY ","

指定列分隔符，默认为\t。

columns_mapping

(k1,k2,tmpk1,k3=tmpk1+1)

指定文件列和表中列的映射关系，以及各种列转换等。详细说明请参见[导入的数据转换、列映射及过滤]。

preceding_filter

无

指定过滤原始数据条件。详细说明请参见[导入的数据转换、列映射及过滤]。

where_predicates

WHERE k1>100 and k2=1000

指定条件对导入的数据进行过滤。详细说明请参见[导入的数据转换、列映射及过滤]。

partitions

PARTITION(p1,p2,p3)

指定导入目的表的哪些Partition中。如果不指定，则会自动导入到对应的Partition中。

DELETE ON

DELETE ON v3>100

需配合MEREGE导入模式一起使用，仅针对Unique Key模型的表。用于指定导入数据中表示Delete Flag的列和计算关系。

ORDER BY

无

仅针对Unique Key模型的表。用于指定导入数据中表示Sequence Col的列。主要用于导入时保证数据顺序。

job_properties参数说明

PROPERTIES (
    "key1" = "val1",
    "key2" = "val2"
)
说明

参数max_batch_interval、max_batch_rows、max_batch_size，三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。

参数名称

示例值

参数说明

desired_concurrent_number

"desired_concurrent_number" = "3"

指定期望并发度。默认为3。一个例行导入作业会被分成多个子任务执行。这个参数用于指定一个作业最多有多少任务可以同时执行。必须大于0。

说明

这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合判断。

max_batch_interval

"max_batch_interval" = "20"

指定每个子任务最大执行时间，单位是秒，默认为10，取值范围为5~60秒。

max_batch_rows

"max_batch_rows" = "300000"

指定每个子任务最多读取的行数。默认是200000，取值范围大于等于200000。

max_batch_size

"max_batch_size" = "209715200"

指定每个子任务最多读取的字节数。单位是字节，默认为104857600，即100 MB。取值范围为100 MB~1 GB。

max_error_number

"max_error_number"="3"

指定采样窗口内，允许的最大错误行数。默认为0，即不允许有错误行。取值范围大于等于0。

采样窗口为max_batch_rows*10。即如果在采样窗口内，错误行数大于，则会导致例行作业被暂停，需要人工介入检查数据质量问题。

说明

被where条件过滤掉的行不算错误行。

strict_mode

"strict_mode"="true"

指定是否开启严格模式，默认为false。开启后，非空原始数据的列类型变换如果结果为NULL，则会被过滤。指定方式为严格模式时，即对于导入过程中的列类型转换进行严格过滤。严格过滤的策略如下：

  • 对于列类型转换来说，如果strict mode为true，则错误的数据将被过滤。错误数据即原始数据并不为空值，在参与列类型转换后结果为空值的这一类数据。

  • 对于导入的某列由函数变换生成时，strict mode对其不产生影响。

  • 对于导入的某列类型包含范围限制的，如果原始数据能正常通过类型转换，但无法通过范围限制的，strict mode对其也不产生影响。例如：如果类型是decimal(1,0)，原始数据为10，则属于可以通过类型转换但不在列声明的范围内。这种数据strict对其不产生影响。

timezone

"timezone" = "Africa/Abidjan"

指定导入作业所使用的时区。默认为使用Session的时区作为参数。

说明

该参数会影响所有导入涉及的和时区有关的函数结果。

format

"format" = "json"

指定导入数据格式，默认为CSV，支持JSON格式。

jsonpaths

-H "jsonpaths:[\"$.k2\",\"$.k1\"]"

当导入数据格式为JSON时，可通过jsonpaths指定抽取JSON数据中的字段。

strip_outer_array

-H "strip_outer_array:true"

当导入数据格式为JSON时，strip_outer_array为true表示JSON数据以数组的形式展现，数据中的每一个元素将被视为一行数据。默认为false。

json_root

-H "json_root:$.RECORDS"

当导入数据格式为JSON时，可以通过json_root指定JSON数据的根节点。SelectDB将通过json_root抽取根节点的元素进行解析。默认为空。

send_batch_parallelism

无

指定设置发送批处理数据的并行度，如果并行度的值超过BE配置中的max_send_batch_parallelism_per_job，那么作为协调点的BE将使用max_send_batch_parallelism_per_job的值。

load_to_single_tablet

无

指定是否只导入数据到对应分区的一个tablet，默认值为false。该参数只允许向对带有random分区的Duplicate表导入数据的时设置。

严格模式（strict mode）与原始数据（source data）的导入关系

例如列类型为TinyInt。当表中的列允许导入空值时导入关系如下。

source data

source data example

string to int

strict_mode

result

空值

\N

N/A

true or false

NULL

not null

aaa or 2000

NULL

true

invalid data(filtered)

not null

aaa

NULL

false

NULL

not null

1

1

true or false

correct data

例如列类型为Decimal(1,0)。当表中的列允许导入空值时导入关系如下。

source data

source data example

string to int

strict_mode

result

空值

\N

N/A

true or false

NULL

not null

aaa

NULL

true

invalid data(filtered)

not null

aaa

NULL

false

NULL

not null

1 or 10

1

true or false

correct data

说明

10虽然是一个超过范围的值，但是因为其类型符合Decimal的要求，所以strict mode对其不产生影响。10最后会在其他ETL处理流程中被过滤。但不会被strict mode过滤。

data_source_properties参数说明

FROM KAFKA
(
    "key1" = "val1",
    "key2" = "val2"
)

参数名称

参数说明

kafka_broker_list

指定Kafka的Broker连接信息。格式为ip:host。多个Broker之间以逗号分隔。

格式："kafka_broker_list"="broker1:9092,broker2:9092"。

kafka_topic

指定订阅的Kafka的Topic。

格式："kafka_topic"="my_topic"。

kafka_partitions/kafka_offsets

指定订阅的Kafka Partition，以及对应的每个Partition的起始Offset。如果指定时间，则会从大于等于该时间的最近一个Offset处开始消费。

Offset可以指定从大于等于0的具体Offset，或者：

  • OFFSET_BEGINNING：从有数据的位置开始订阅。

  • OFFSET_END：从末尾开始订阅。

  • 时间格式，如："2021-05-22 11:00:00"

如果没有指定，则默认从OFFSET_END开始订阅Topic下的所有Partition。

示例如下。

"kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"
"kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "2021-05-22 11:00:00,2021-05-22 11:00:00,2021-05-22 11:00:00"
重要

时间格式不能和OFFSET格式混用。

property

指定自定义Kafka参数。功能等同于Kafka shell中"--property"参数。

当参数的value为一个文件时，需要在value前加上关键词："FILE:"。

Property参数说明

  • 使用SSL连接Kafka时，需要指定以下参数：

    "property.security.protocol" = "ssl",
"property.ssl.ca.location" = "FILE:ca.pem",
"property.ssl.certificate.location" = "FILE:client.pem",
"property.ssl.key.location" = "FILE:client.key",
"property.ssl.key.password" = "abcdefg"

    其中property.security.protocol和property.ssl.ca.location为必选项，用于指明连接方式为SSL，以及CA证书的位置。

    如果Kafka Server端开启了Client认证，则还需设置以下参数。

    "property.ssl.certificate.location"
"property.ssl.key.location"
"property.ssl.key.password"

    分别用于指定Client的public key、private key、private key的密码。

  • 指定Kafka Partition的默认起始offset。

    没有指定kafka_partitions/kafka_offsets，默认消费所有分区。此时可以通过kafka_default_offsets指定起始offset。默认为OFFSET_END，即从末尾开始订阅。

    "property.kafka_default_offsets" = "OFFSET_BEGINNING"

更多支持的自定义参数，请参阅librdkafka的官方CONFIGURATION文档中，Client端的配置项。例如以下参数。

"property.client.id" = "12345",
"property.ssl.ca.location" = "FILE:ca.pem"

使用示例

创建Routine Load简单任务

  1. 创建待导入的SelectDB数据表，示例如下。

    CREATE TABLE test_table
(
    id int,
    name varchar(50),
    age int,
    address varchar(50),
    url varchar(500)
)
UNIQUE KEY(`id`, `name`)
DISTRIBUTED BY HASH(id) BUCKETS 4
PROPERTIES("replication_num" = "1");

  2. 分别设置不同的参数导入数据，示例如下。

    • 为example_db的test_table创建一个名为test1的Kafka Routine Load任务。指定列分隔符的group.id和client.id，设置自动默认消费所有分区，且从有数据的位置（OFFSET_BEGINNING）开始订阅，示例如下。

      CREATE ROUTINE LOAD example_db.test1 ON test_table
COLUMNS TERMINATED BY ",",
COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

    • 为example_db的test_table创建一个名为test2的Kafka Routine Load任务。导入任务为严格模式，示例如下。

      CREATE ROUTINE LOAD example_db.test2 ON test_table
COLUMNS TERMINATED BY ",",
COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "true"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

    • 从指定的时间点开始消费，示例如下。

      CREATE ROUTINE LOAD example_db.test4 ON test_table
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "30",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200"
) FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092",
    "kafka_topic" = "my_topic",
    "property.kafka_default_offset" = "2024-01-21 10:00:00"
);

导入JSON格式数据

Routine Load导入的JSON格式仅支持以下两种。

  • 只有一条记录，且为JSON对象。

    当使用单表导入（即通过ON TABLE_NAME指定表名）时，JSON数据格式如下。

    {"key1":"value1","key2":"value2","key3":"value3"}

    当使用动态或多表导入Routine Load（即不指定具体的表名）时，JSON数据格式如下。

    table_name|{"key1":"value1","key2":"value2","key3":"value3"}

  • JSON数组，数组中可含多条记录。

    当使用单表导入（即通过ON TABLE_NAME指定表名）时，JSON数据格式如下。

    [
    {   
        "key1":"value11",
        "key2":"value12",
        "key3":"value13",
        "key4":14
    },
    {
        "key1":"value21",
        "key2":"value22",
        "key3":"value23",
        "key4":24
    },
    {
        "key1":"value31",
        "key2":"value32",
        "key3":"value33",
        "key4":34
    }
]

    当使用动态或多表导入（即不指定具体的表名）时，JSON数据格式如下。

       table_name|[
    {   
        "key1":"value11",
        "key2":"value12",
        "key3":"value13",
        "key4":14
    },
    {
        "key1":"value21",
        "key2":"value22",
        "key3":"value23",
        "key4":24
    },
    {
        "key1":"value31",
        "key2":"value32",
        "key3":"value33",
        "key4":34
    }
]

导入JSON格式数据，示例如下。

  1. 创建待导入的SelectDB数据表，示例如下。

    CREATE TABLE `example_tbl` (
   `category` varchar(24) NULL COMMENT "",
   `author` varchar(24) NULL COMMENT "",
   `timestamp` bigint(20) NULL COMMENT "",
   `dt` int(11) NULL COMMENT "",
   `price` double REPLACE
) ENGINE=OLAP
AGGREGATE KEY(`category`,`author`,`timestamp`,`dt`)
COMMENT "OLAP"
PARTITION BY RANGE(`dt`)
(
  PARTITION p0 VALUES [("-2147483648"), ("20230509")),
    PARTITION p20200509 VALUES [("20230509"), ("20231010")),
    PARTITION p20200510 VALUES [("20231010"), ("20231211")),
    PARTITION p20200511 VALUES [("20231211"), ("20240512"))
)
DISTRIBUTED BY HASH(`category`,`author`,`timestamp`) BUCKETS 4;

  2. 分别使用两个类型的JSON数据记录到Topic里：

    {
    "category":"value1331",
    "author":"value1233",
    "timestamp":1700346050,
    "price":1413
}
    [
    {
        "category":"value13z2",
        "author":"vaelue13",
        "timestamp":1705645251,
        "price":14330
    },
    {
        "category":"lvalue211",
        "author":"lvalue122",
        "timestamp":1684448450,
        "price":24440
    }
]

  3. 以不同模式导入JSON数据，示例如下。

    • 以简单模式导入JSON数据。

      CREATE ROUTINE LOAD example_db.test_json_label_1 ON example_tbl
COLUMNS(category,price,author)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
 );

    • 精准导入JSON格式数据。

      CREATE ROUTINE LOAD example_db.test_json_label_3 ON example_tbl
COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json",
    "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
    "strip_outer_array" = "true"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
);
说明

表里的分区字段dt在示例数据里并没有，而是在Routine Load语句里通过dt=from_unixtime(timestamp,'%Y%m%d')转换出来的。

访问不同验证方式的Kafka集群

根据Kafka集群验证方式的不同，访问的方式示例如下。

  • 访问SSL认证的Kafka集群。

    访问SSL认证的Kafka集群需要您提供用于认证Kafka Broker公钥的证书文件（ca.pem）。如果Kafka集群同时开启了客户端认证，则还需提供客户端的公钥（client.pem）、密钥文件（client.key），以及密钥密码。这里所需的文件需要先通过CREATE FILE命令上传到SelectDB中，并且Catalog名称为kafka。

    1. 上传文件，示例如下。

      CREATE FILE "ca.pem" PROPERTIES("url" = "https://example_url/kafka-key/ca.pem", "catalog" = "kafka");
CREATE FILE "client.key" PROPERTIES("url" = "https://example_urlkafka-key/client.key", "catalog" = "kafka");
CREATE FILE "client.pem" PROPERTIES("url" = "https://example_url/kafka-key/client.pem", "catalog" = "kafka");

    2. 创建Routine Load作业，示例如下。

      CREATE ROUTINE LOAD db1.job1 on tbl1
PROPERTIES
(
    "desired_concurrent_number"="1"
)
FROM KAFKA
(
    "kafka_broker_list"= "broker1:9091,broker2:9091",
    "kafka_topic" = "my_topic",
    "property.security.protocol" = "ssl",
    "property.ssl.ca.location" = "FILE:ca.pem",
    "property.ssl.certificate.location" = "FILE:client.pem",
    "property.ssl.key.location" = "FILE:client.key",
    "property.ssl.key.password" = "abcdefg"
);
      说明

      SelectDB通过Kafka的C++ APIlibrdkafka来访问Kafka集群。librdkafka所支持的参数可以参阅Configuration properties。

  • 访问PLAIN认证的Kafka集群。

    访问开启PLAIN认证的Kafka集群，需要增加配置如下。

    • property.security.protocol=SASL_PLAINTEXT：使用SASL plaintext。

    • property.sasl.mechanism=PLAIN：设置SASL的认证方式为PLAIN。

    • property.sasl.username=admin：设置SASL的用户名。

    • property.sasl.password=admin：设置SASL的密码。

    创建Routine Load作业，示例如下。

    CREATE ROUTINE LOAD db1.job1 on tbl1
PROPERTIES (
"desired_concurrent_number"="1",
 )
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092",
    "kafka_topic" = "my_topic",
    "property.security.protocol"="SASL_PLAINTEXT",
    "property.sasl.mechanism"="PLAIN",
    "property.sasl.username"="admin",
    "property.sasl.password"="admin"
);

  • 访问Kerberos认证的Kafka集群。

    访问开启kerberos认证的Kafka集群，需要增加配置如下。

    • security.protocol=SASL_PLAINTEXT：使用SASL plaintext。

    • sasl.kerberos.service.name=$SERVICENAME：设置broker servicename。

    • sasl.kerberos.keytab=/etc/security/keytabs/${CLIENT_NAME}.keytab：设置Keytab本地文件路径。

    • sasl.kerberos.principal=${CLIENT_NAME}/${CLIENT_HOST}：设置SelectDB连接Kafka时使用的Kerberos主体。

    创建Routine Load作业，示例如下。

    CREATE ROUTINE LOAD db1.job1 on tbl1
PROPERTIES (
"desired_concurrent_number"="1",
 )
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092",
    "kafka_topic" = "my_topic",
    "property.security.protocol" = "SASL_PLAINTEXT",
    "property.sasl.kerberos.service.name" = "kafka",
    "property.sasl.kerberos.keytab" = "/etc/krb5.keytab",
    "property.sasl.kerberos.principal" = "id@your.com"
);
说明

  • 若要使SelectDB访问开启kerberos认证方式的Kafka集群，需要在SelectDB集群所有运行节点上部署Kerberos客户端kinit，并配置krb5.conf，填写KDC服务信息等。

  • 配置property.sasl.kerberos.keytab的值需要指定Keytab本地文件的绝对路径，并允许SelectDB进程访问该本地文件。

修改导入

修改已经创建的例行导入作业。只能修改处于PAUSED状态的作业。

语法

ALTER ROUTINE LOAD FOR <job_name>
[job_properties]
FROM <data_source>
[data_source_properties]

参数说明

参数名称

参数说明

[db.]job_name

指定要修改的作业名称。

tbl_name

指定需要导入的表的名称。

job_properties

指定需要修改的作业参数。目前仅支持修改的参数如下。

  • desired_concurrent_number。

  • max_error_number。

  • max_batch_interval。

  • max_batch_rows。

  • max_batch_size。

  • jsonpaths。

  • json_root。

  • strip_outer_array。

  • strict_mode。

  • timezone。

  • num_as_string。

  • fuzzy_parse。

data_source

指定数据源的类型。当前支持：KAFKA。

data_source_properties

指定数据源的相关属性。目前仅支持的属性如下。

  1. kafka_partitions。

  2. kafka_offsets。

  3. kafka_broker_list。

  4. kafka_topic。

  5. 自定义property，如property.group.id。

说明

kafka_partitions和kafka_offsets用于修改待消费的kafka partition 的offset，仅能修改当前已经消费的partition。不能新增partition。

使用示例

  • 将desired_concurrent_number修改为1，示例如下。

    ALTER ROUTINE LOAD FOR db1.label1
PROPERTIES
(
    "desired_concurrent_number" = "1"
);

  • 将desired_concurrent_number修改为10，修改partition的offset，修改group id，示例如下。

    ALTER ROUTINE LOAD FOR db1.label1
PROPERTIES
(
    "desired_concurrent_number" = "10"
)
FROM kafka
(
    "kafka_partitions" = "0, 1, 2",
    "kafka_offsets" = "100, 200, 100",
    "property.group.id" = "new_group"
);

暂停导入

暂停一个Routine Load作业。被暂停的作业可以通过RESUME命令重新运行。

语法

PAUSE [ALL] ROUTINE LOAD FOR <job_name>;

参数说明

参数名称

参数说明

[db.]job_name

指定要暂停的作业名称。

使用示例

  • 暂停名称为test1的Routine Load作业，示例如下。

    PAUSE ROUTINE LOAD FOR test1;

  • 暂停所有Routine Load作业，示例如下。

    PAUSE ALL ROUTINE LOAD;

恢复导入

恢复一个被暂停的Routine Load作业。恢复的作业将继续从之前已消费的offset继续消费。

语法

RESUME [ALL] ROUTINE LOAD FOR <job_name>

参数说明

参数名称

参数说明

[db.]job_name

指定要恢复的作业名称。

使用示例

  • 恢复名称为test1的Routine Load作业，示例如下。

    RESUME ROUTINE LOAD FOR test1;

  • 恢复所有Routine Load作业，示例如下。

    RESUME ALL ROUTINE LOAD;

停止导入

停止一个Routine Load作业。被停止的作业无法再重新运行。停止导入后，已导入数据不会回滚。

语法

STOP ROUTINE LOAD FOR <job_name>;

参数说明

参数名称

参数说明

[db.]job_name

指定要停止的作业名称。

使用示例

停止名称为test1的Routine Load作业，示例如下。

STOP ROUTINE LOAD FOR test1;

查看导入

Routine Load作业运行状态需要通过SHOW ROUTINE LOAD命令查看。

语法

SHOW [ALL] ROUTINE LOAD [FOR job_name];

参数说明

参数名称

参数说明

[db.]job_name

指定要查看的作业名称。

说明

如果导入数据格式不合法，详细的错误信息会记录在ErrorLogUrls中。注意其中包含多个链接，拷贝其中任意一个链接在浏览器中查看即可。

使用示例

  • 展示名称为test1的所有Routine Load作业（包括已停止或取消的作业）。结果为一行或多行。

    SHOW ALL ROUTINE LOAD FOR test1;

  • 展示名称为test1的当前正在运行的Routine Load作业。

    SHOW ROUTINE LOAD FOR test1;

  • 显示example_db下，所有的Routine Load作业（包括已停止或取消的作业）。结果为一行或多行。

    use example_db;
SHOW ALL ROUTINE LOAD;

  • 显示example_db下，所有正在运行的Routine Load作业。

    use example_db;
SHOW ROUTINE LOAD;

  • 显示example_db下，名称为test1的当前正在运行的Routine Load作业。

    SHOW ROUTINE LOAD FOR example_db.test1;

  • 显示example_db下，名称为test1的所有Routine Load作业（包括已停止或取消的作业）。结果为一行或多行。

    SHOW ALL ROUTINE LOAD FOR example_db.test1;

相关系统配置​

相关系统配置参数会影响Routine Load的使用。

  • max_routine_load_task_concurrent_num

    FE配置项，默认为5，可以运行时修改。该参数限制了一个例行导入作业最大的子任务并发数。建议维持默认值。设置过大，可能导致同时并发的任务数过多，占用集群资源。

  • max_routine_load_task_num_per_be

    FE配置项，默认为5，可以运行时修改。该参数限制了每个BE节点最多并发执行的子任务个数。建议维持默认值。如果设置过大，可能导致并发任务数过多，占用集群资源。

  • max_routine_load_job_num

    FE配置项，默认为100，可以运行时修改。该参数限制Routine Load作业的总数，包括NEED_SCHEDULED，RUNNING，PAUSE这些状态。超过后，不能再提交新的作业。

  • max_consumer_num_per_group

    BE 配置项，默认为3。该参数表示一个子任务中最多生成几个Consumer进行数据消费。对于Kafka数据源，一个Consumer可能消费一个或多个Kafka Partition。假设一个任务需要消费6个Kafka Partition，则会生成3个Consumer，每个Consumer消费2个Partition。如果只有2个Partition，则只会生成2个Consumer，每个Consumer消费1个Partition。

  • max_tolerable_backend_down_num

    FE配置项，默认值是0。在满足某些条件下，SelectDB可令PAUSED的任务重新调度，变成RUNNING。该参数为0代表只有所有BE节点是alive状态才允许重新调度。

  • period_of_auto_resume_min

    FE配置项，默认是5分钟。该项意味着当SelectDB重新调度任务时，只会在5分钟这个周期内最多尝试3次。如果3次都失败则锁定当前任务，后续不再进行调度，但可通过人为干预，进行手动恢复。

其他说明​

  • Routine Load作业和ALTER TABLE操作的关系。

    • Routine Load不会阻塞Schema变更和ROLLUP操作。但Schema变更完成后，列映射关系无法匹配，则会导致作业的错误数据激增，最终导致作业暂停。建议通过在Routine Load作业中显式指定列映射关系，并且通过增加Nullable列或带Default值的列来减少这些问题。

    • 删除表的Partition可能会导致导入数据无法找到对应的Partition，导致作业进入暂停状态。

  • Routine Load作业和其他导入作业的关系（LOAD，DELETE，INSERT）。

    • Routine Load和其他LOAD作业以及INSERT操作没有冲突。

    • 当执行DELETE操作时，对应表分区不能有任何正在执行的导入任务。因此在执行DELETE操作前，需要暂停Routine Load作业，并等待已下发的任务全部完成后，方可以执行DELETE。

  • Routine Load作业和DROP DATABASE或DROP TABLE操作的关系。

    当Routine Load对应的database或table被删除后，作业会自动CANCEL。

  • Kafka类型的Routine Load作业和Kafka topic的关系。

    当创建例行导入声明的kafka_topic在Kafka集群中不存在时：

    • Kafka集群的Broker设置了auto.create.topics.enable=true，则topic会先被自动创建，自动创建的Partition个数是由您的Kafka集群中的Broker配置num.partitions决定的。例行作业会不断读取该topic的数据。

    • Kafka集群的Broker设置了auto.create.topics.enable=false,则topic不会被自动创建，例行作业会在没有读取任何数据之前就被暂停，状态置为PAUSED。

    所以，如果您希望当Kafka topic不存在的时候，被例行作业自动创建的话，只需要将Kafka集群中的Broker设置auto.create.topics.enable=true即可。

  • 当环境中存在网段和域名解析的隔离措施，因此需要注意以下问题。

    • 创建Routine Load任务中指定的Broker list必须能够被SelectDB服务访问。

    • Kafka中如果配置了advertised.listeners,advertised.listeners中的地址必须能够被SelectDB服务访问。

  • 指定消费的Partition和Offset。

    SelectDB支持指定Partition，Offset和时间点进行消费。这里说明下对应参数的配置关系。

    三个相关参数如下。

    • kafka_partitions：指定待消费的Partition列表，如："0,1,2,3"。

    • kafka_offsets：指定每个分区的起始offset，必须和kafka_partitions列表个数对应。如："1000,1000,2000,2000"

    • property.kafka_default_offset：指定分区默认的起始Offset。

    在创建导入作业时，这三个参数可以有以下五种组合方式。

    方式

    kafka_partitions

    kafka_offsets

    property.kafka_default_offset

    行为

    1

    No

    No

    No

    系统会自动查找Topic对应的所有分区并从OFFSET_END开始消费。

    2

    No

    No

    Yes

    系统会自动查找Topic对应的所有分区并从default offset指定的位置开始消费。

    3

    Yes

    No

    No

    系统会从指定分区的OFFSET_END开始消费。

    4

    Yes

    Yes

    No

    系统会从指定分区的指定Offset处开始消费。

    5

    Yes

    No

    Yes

    系统会从指定分区，default Offset指定的位置开始消费。

  • STOP和PAUSE的区别。

    FE会自动定期清理STOP状态的Routine Load，而PAUSE状态的则可以再次被恢复启用。

文档推荐
  • 本页导读 （1）
文档反馈