ROUTINE LOAD

description

1. 例行导入（Routine Load）功能，支持用户提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 Doris 中。
2. 目前仅支持通过无认证或者 SSL 认证方式，从 Kakfa 导入文本格式（CSV）的数据。

语法：

1. CREATE ROUTINE LOAD [db.]job_name ON tbl_name
2. [merge_type]
3. [load_properties]
4. [job_properties]
5. FROM data_source
6. [data_source_properties]

8. 1. [db.]job_name

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

12. 2. tbl_name

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

16. 3. merge_type
17. 数据的合并类型，一共支持三种类型APPEND、DELETE、MERGE 其中，APPEND是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据key相同的所有行，MERGE 语义 需要与delete on条件联合使用，表示满足delete 条件的数据按照DELETE 语义处理其余的按照APPEND 语义处理, 语法为[WITH MERGE|APPEND|DELETE]

19. 4. load_properties

21. 用于描述导入数据。语法：
22. [column_separator],
23. [columns_mapping],
24. [where_predicates],
25. [delete_on_predicates],
26. [source_sequence],
27. [partitions],
28. [preceding_predicates]

30. 1. column_separator:

32. 指定列分隔符，如：

34. COLUMNS TERMINATED BY ","

36. 默认为：\t

38. 2. columns_mapping:

40. 指定源数据中列的映射关系，以及定义衍生列的生成方式。

42. 1. 映射列：

44. 按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。
45. 假设目的表有三列 k1, k2, v1。源数据有4列，其中第1、2、4列分别对应 k2, k1, v1。则书写如下：

47. COLUMNS (k2, k1, xxx, v1)

49. 其中 xxx 为不存在的一列，用于跳过源数据中的第三列。

51. 2. 衍生列：

53. 以 col_name = expr 的形式表示的列，我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。
54. 衍生列通常排列在映射列之后，虽然这不是强制的规定，但是 Doris 总是先解析映射列，再解析衍生列。
55. 接上一个示例，假设目的表还有第4列 v2，v2 由 k1 和 k2 的和产生。则可以书写如下：

57. COLUMNS (k2, k1, xxx, v1, v2 = k1 + k2);

59. 3. where_predicates

61. 用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。
62. 例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：

64. WHERE k1 > 100 and k2 = 1000

66. 4. partitions

68. 指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。
69. 示例：

71. PARTITION(p1, p2, p3)

73. 5. delete_on_predicates
74. 表示删除条件，仅在 merge type 为MERGE 时有意义，语法与where 相同

76. 6. source_sequence:

78. 只适用于UNIQUE_KEYS,相同key列下，保证value列按照source_sequence列进行REPLACE, source_sequence可以是数据源中的列，也可以是表结构中的一列。

80. 7. preceding_predicates

82. PRECEDING FILTER predicate

84. 用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。

86. 5. job_properties

88. 用于指定例行导入作业的通用参数。
89. 语法：

91. PROPERTIES (
92. "key1" = "val1",
93. "key2" = "val2"
94. )

96. 目前我们支持以下参数：

98. 1. desired_concurrent_number

100. 期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于0。默认为3。
101. 这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。
102. 例：

104. "desired_concurrent_number" = "3"

106. 2. max_batch_interval/max_batch_rows/max_batch_size

108. 这三个参数分别表示：
109. 1）每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为10。
110. 2）每个子任务最多读取的行数。必须大于等于200000。默认是200000。
111. 3）每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是 100MB。

113. 这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。
114. 例：

116. "max_batch_interval" = "20",
117. "max_batch_rows" = "300000",
118. "max_batch_size" = "209715200"

120. 3. max_error_number

122. 采样窗口内，允许的最大错误行数。必须大于等于0。默认是 0，即不允许有错误行。
123. 采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。
124. 被 where 条件过滤掉的行不算错误行。

126. 4. strict_mode

128. 是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为 NULL，则会被过滤。指定方式为 "strict_mode" = "true"

130. 5. timezone

132. 指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果。

134. 6. format

136. 指定导入数据格式，默认是csv，支持json格式。

138. 7. jsonpaths

140. jsonpaths: 导入json方式分为：简单模式和匹配模式。如果设置了jsonpath则为匹配模式导入，否则为简单模式导入，具体可参考示例。

142. 8. strip_outer_array

144. 布尔类型，为true表示json数据以数组对象开始且将数组对象中进行展平，默认值是false。

146. 9. json_root

148. json_root为合法的jsonpath字符串，用于指定json document的根节点，默认值为""。

150. 10. send_batch_parallelism

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

154. 6. data_source

156. 数据源的类型。当前支持：

158. KAFKA

160. 7. data_source_properties

162. 指定数据源相关的信息。
163. 语法：

165. (
166. "key1" = "val1",
167. "key2" = "val2"
168. )

170. 1. KAFKA 数据源

172. 1. kafka_broker_list

174. Kafka 的 broker 连接信息。格式为 ip:host。多个broker之间以逗号分隔。
175. 示例：

177. "kafka_broker_list" = "broker1:9092,broker2:9092"

179. 2. kafka_topic

181. 指定要订阅的 Kafka 的 topic。
182. 示例：

184. "kafka_topic" = "my_topic"

186. 3. kafka_partitions/kafka_offsets

188. 指定需要订阅的 kafka partition，以及对应的每个 partition 的起始 offset。

190. offset 可以指定从大于等于 0 的具体 offset，或者：
191. 1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
192. 2) OFFSET_END: 从末尾开始订阅。
193. 3) 时间戳，格式必须如："2021-05-11 10:00:00"，系统会自动定位到大于等于该时间戳的第一个消息的offset。注意，时间戳格式的offset不能和数字类型混用，只能选其一。

195. 如果没有指定，则默认从 OFFSET_END 开始订阅 topic 下的所有 partition。
196. 示例：

198. "kafka_partitions" = "0,1,2,3",
199. "kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"

201. "kafka_partitions" = "0,1",
202. "kafka_offsets" = "2021-05-11 10:00:00, 2021-05-11 11:00:00"

204. 4. property

206. 指定自定义kafka参数。
207. 功能等同于kafka shell中 "--property" 参数。
208. 当参数的 value 为一个文件时，需要在 value 前加上关键词："FILE:"。
209. 关于如何创建文件，请参阅 "HELP CREATE FILE;"
210. 更多支持的自定义参数，请参阅 librdkafka 的官方 CONFIGURATION 文档中，client 端的配置项。

212. 示例:
213. "property.client.id" = "12345",
214. "property.ssl.ca.location" = "FILE:ca.pem"

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

218. "property.security.protocol" = "ssl",
219. "property.ssl.ca.location" = "FILE:ca.pem",
220. "property.ssl.certificate.location" = "FILE:client.pem",
221. "property.ssl.key.location" = "FILE:client.key",
222. "property.ssl.key.password" = "abcdefg"

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

227. 如果 Kafka server 端开启了 client 认证，则还需设置：

229. "property.ssl.certificate.location"
230. "property.ssl.key.location"
231. "property.ssl.key.password"

233. 分别用于指定 client 的 public key，private key 以及 private key 的密码。

236. 2.指定kafka partition的默认起始offset
237. 如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。
238. 值为
239. 1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
240. 2) OFFSET_END: 从末尾开始订阅。
241. 3) 时间戳，格式同 kafka_offsets

243. 示例：
244. "property.kafka_default_offsets" = "OFFSET_BEGINNING"
245. "property.kafka_default_offsets" = "2021-05-11 10:00:00"

247. 8. 导入数据格式样例

249. 整型类（TINYINT/SMALLINT/INT/BIGINT/LARGEINT）：1, 1000, 1234
250. 浮点类（FLOAT/DOUBLE/DECIMAL）：1.1, 0.23, .356
251. 日期类（DATE/DATETIME）：2017-10-03, 2017-06-13 12:34:03。
252. 字符串类（CHAR/VARCHAR）（无引号）：I am a student, a
253. NULL值：\N

example

1. 1. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。指定列分隔符和 group.id 和 client.id，并且自动默认消费所有分区，且从有数据的位置（OFFSET_BEGINNING）开始订阅

3. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
4. COLUMNS TERMINATED BY ",",
5. COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
6. PROPERTIES
7. (
8. "desired_concurrent_number"="3",
9. "max_batch_interval" = "20",
10. "max_batch_rows" = "300000",
11. "max_batch_size" = "209715200",
12. "strict_mode" = "false"
13. )
14. FROM KAFKA
15. (
16. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
17. "kafka_topic" = "my_topic",
18. "property.group.id" = "xxx",
19. "property.client.id" = "xxx",
20. "property.kafka_default_offsets" = "OFFSET_BEGINNING"
21. );

23. 2. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。导入任务为严格模式。

25. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
26. COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
27. WHERE k1 > 100 and k2 like "%doris%"
28. PROPERTIES
29. (
30. "desired_concurrent_number"="3",
31. "max_batch_interval" = "20",
32. "max_batch_rows" = "300000",
33. "max_batch_size" = "209715200",
34. "strict_mode" = "false"
35. )
36. FROM KAFKA
37. (
38. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
39. "kafka_topic" = "my_topic",
40. "kafka_partitions" = "0,1,2,3",
41. "kafka_offsets" = "101,0,0,200"
42. );

44. 3. 通过 SSL 认证方式，从 Kafka 集群导入数据。同时设置 client.id 参数。导入任务为非严格模式，时区为 Africa/Abidjan

46. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
47. COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
48. WHERE k1 > 100 and k2 like "%doris%"
49. PROPERTIES
50. (
51. "desired_concurrent_number"="3",
52. "max_batch_interval" = "20",
53. "max_batch_rows" = "300000",
54. "max_batch_size" = "209715200",
55. "strict_mode" = "false",
56. "timezone" = "Africa/Abidjan"
57. )
58. FROM KAFKA
59. (
60. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
61. "kafka_topic" = "my_topic",
62. "property.security.protocol" = "ssl",
63. "property.ssl.ca.location" = "FILE:ca.pem",
64. "property.ssl.certificate.location" = "FILE:client.pem",
65. "property.ssl.key.location" = "FILE:client.key",
66. "property.ssl.key.password" = "abcdefg",
67. "property.client.id" = "my_client_id"
68. );
69. 4. 简单模式导入json
70. CREATE ROUTINE LOAD example_db.test_json_label_1 ON table1
71. COLUMNS(category,price,author)
72. PROPERTIES
73. (
74. "desired_concurrent_number"="3",
75. "max_batch_interval" = "20",
76. "max_batch_rows" = "300000",
77. "max_batch_size" = "209715200",
78. "strict_mode" = "false",
79. "format" = "json"
80. )
81. FROM KAFKA
82. (
83. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
84. "kafka_topic" = "my_topic",
85. "kafka_partitions" = "0,1,2",
86. "kafka_offsets" = "0,0,0"
87. );
88. 支持两种json数据格式：
89. 1）{"category":"a9jadhx","author":"test","price":895}
90. 2）[
91. {"category":"a9jadhx","author":"test","price":895},
92. {"category":"axdfa1","author":"EvelynWaugh","price":1299}
93. ]

95. 5. 精准导入json数据格式
96. CREATE TABLE `example_tbl` (
97. `category` varchar(24) NULL COMMENT "",
98. `author` varchar(24) NULL COMMENT "",
99. `timestamp` bigint(20) NULL COMMENT "",
100. `dt` int(11) NULL COMMENT "",
101. `price` double REPLACE
102. ) ENGINE=OLAP
103. AGGREGATE KEY(`category`,`author`,`timestamp`,`dt`)
104. COMMENT "OLAP"
105. PARTITION BY RANGE(`dt`)
106. (PARTITION p0 VALUES [("-2147483648"), ("20200509")),
107. PARTITION p20200509 VALUES [("20200509"), ("20200510")),
108. PARTITION p20200510 VALUES [("20200510"), ("20200511")),
109. PARTITION p20200511 VALUES [("20200511"), ("20200512")))
110. DISTRIBUTED BY HASH(`category`,`author`,`timestamp`) BUCKETS 4
111. PROPERTIES (
112. "replication_num" = "1"
113. );

115. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
116. COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
117. PROPERTIES
118. (
119. "desired_concurrent_number"="3",
120. "max_batch_interval" = "20",
121. "max_batch_rows" = "300000",
122. "max_batch_size" = "209715200",
123. "strict_mode" = "false",
124. "format" = "json",
125. "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
126. "strip_outer_array" = "true"
127. )
128. FROM KAFKA
129. (
130. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
131. "kafka_topic" = "my_topic",
132. "kafka_partitions" = "0,1,2",
133. "kafka_offsets" = "0,0,0"
134. );

json数据格式: [ {“category”:”11”,”title”:”SayingsoftheCentury”,”price”:895,”timestamp”:1589191587}, {“category”:”22”,”author”:”2avc”,”price”:895,”timestamp”:1589191487}, {“category”:”33”,”author”:”3avc”,”title”:”SayingsoftheCentury”,”timestamp”:1589191387} ] 说明： 1）如果json数据是以数组开始，并且数组中每个对象是一条记录，则需要将strip_outer_array设置成true，表示展平数组。 2）如果json数据是以数组开始，并且数组中每个对象是一条记录，在设置jsonpath时，我们的ROOT节点实际上是数组中对象。

1. 6. 用户指定根节点json_root
2. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
3. COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
4. PROPERTIES
5. (
6. "desired_concurrent_number"="3",
7. "max_batch_interval" = "20",
8. "max_batch_rows" = "300000",
9. "max_batch_size" = "209715200",
10. "strict_mode" = "false",
11. "format" = "json",
12. "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
13. "strip_outer_array" = "true",
14. "json_root" = "$.RECORDS"
15. )
16. FROM KAFKA
17. (
18. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
19. "kafka_topic" = "my_topic",
20. "kafka_partitions" = "0,1,2",
21. "kafka_offsets" = "0,0,0"
22. );

json数据格式: { “RECORDS”:[ {“category”:”11”,”title”:”SayingsoftheCentury”,”price”:895,”timestamp”:1589191587}, {“category”:”22”,”author”:”2avc”,”price”:895,”timestamp”:1589191487}, {“category”:”33”,”author”:”3avc”,”title”:”SayingsoftheCentury”,”timestamp”:1589191387} ] }

1. 7. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。并且删除与v3 >100 行相匹配的key列的行

3. CREATE ROUTINE LOAD example_db.test1 ON example_tbl
4. WITH MERGE
5. COLUMNS(k1, k2, k3, v1, v2, v3),
6. WHERE k1 > 100 and k2 like "%doris%",
7. DELETE ON v3 >100
8. PROPERTIES
9. (
10. "desired_concurrent_number"="3",
11. "max_batch_interval" = "20",
12. "max_batch_rows" = "300000",
13. "max_batch_size" = "209715200",
14. "strict_mode" = "false"
15. )
16. FROM KAFKA

18. 8. 导入数据到含有sequence列的UNIQUE_KEYS表中
19. CREATE ROUTINE LOAD example_db.test_job ON example_tbl
20. COLUMNS TERMINATED BY ",",
21. COLUMNS(k1,k2,source_sequence,v1,v2),
22. ORDER BY source_sequence
23. PROPERTIES
24. (
25. "desired_concurrent_number"="3",
26. "max_batch_interval" = "30",
27. "max_batch_rows" = "300000",
28. "max_batch_size" = "209715200"
29. ) FROM KAFKA
30. (
31. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
32. "kafka_topic" = "my_topic",
33. "kafka_partitions" = "0,1,2,3",
34. "kafka_offsets" = "101,0,0,200"
35. );

37. 8. 过滤原始数据

39. CREATE ROUTINE LOAD example_db.test_job ON example_tbl
40. COLUMNS TERMINATED BY ",",
41. COLUMNS(k1,k2,source_sequence,v1,v2),
42. PRECEDING FILTER k1 > 2
43. PROPERTIES
44. (
45. "desired_concurrent_number"="3",
46. "max_batch_interval" = "30",
47. "max_batch_rows" = "300000",
48. "max_batch_size" = "209715200"
49. ) FROM KAFKA
50. (
51. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
52. "kafka_topic" = "my_topic",
53. "kafka_partitions" = "0,1,2,3",
54. "kafka_offsets" = "101,0,0,200"
55. );

57. 9. 从指定的时间点开始消费

59. CREATE ROUTINE LOAD example_db.test_job ON example_tbl
60. PROPERTIES
61. (
62. "desired_concurrent_number"="3",
63. "max_batch_interval" = "30",
64. "max_batch_rows" = "300000",
65. "max_batch_size" = "209715200"
66. ) FROM KAFKA
67. (
68. "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
69. "kafka_topic" = "my_topic",
70. "property.kafka_default_offsets" = "2021-10-10 11:00:00"
71. );

keyword

1. CREATE,ROUTINE,LOAD