routine load 消费速率达到瓶颈

💬 StarRocks 用户问答 数据导入与导出
#1

【详述】每天数据量200亿左右，经过过滤条件剩下真正导入的数据在20亿左右。kafka topic里分120个分区。建表一个分区分了64个桶。表有80列,是明细表，其中有8列建了bloom 5列建了bitmap索引。

之前起了两个导入任务，每个任务设置13个task。每个任务消费一半分区。 感觉速度不够。就改成四个导入任务，每个任务设置13个task。然后确实快了一倍，但是消费速度还是有点跟不上。最后直接改成
八个导入任务，每个任务设置13个task。但是这次消费速度没有提升。然后我怎么调试参数就都没有再提升速率了(比如将导入时间由2分钟改为1分钟啥的)。

现在kafka监控显示每秒消费13w条左右。感觉起了这么多任务不应该这么低啊?是不是哪儿有问题

现在集群总的导入效率大概为每秒
400mb/s & 50k ops/s （之前还有别的导入任务）

我看了下日志情况，发现data_consumer 每次实际上消费时间都很短，大部分时间都阻塞了，怀疑是file_node的时候解析json成为瓶颈？而且be.INFO里也有大量的rollback日志。不清楚是为什么!

【导入/导出方式】 routine load 导入数据
【背景】配置routineload 导入一次消费时间2分钟
【业务影响】导入速率跟不上
【StarRocks版本】例如：2.1.1
【集群规模】例如：3fe（3 follower）+13be（fe与be混部）
【机器信息】40C/300G/万兆
【附件】

图片_lanxin_20220615181625
图片_lanxin_202206151816251519×156 157 KB

图片_lanxin_20220615181500
图片_lanxin_20220615181500427×774 174 KB

lanxin_20220615151300
lanxin_202206151513001827×691 908 KB

lanxin_20220615151116
lanxin_20220615151116759×708 132 KB

#2

建表语句:
表一共有100列。
下面是建表的具体配置：
INDEX xx1( 1),
INDEX xx2( 2),
INDEX xx3( 3),
INDEX xx4( 4),
INDEX xx5( 5),
INDEX xx6( 6),
INDEX xx7( 7)
) ENGINE=OLAP
DUPLICATE KEY(mid)
PARTITION BY RANGE (event_date) (
START (“2022-05-13”) END (“2022-06-13”) EVERY (INTERVAL 1 day)
)
DISTRIBUTED BY HASH(mid) BUCKETS 48
PROPERTIES(
“bloom_filter_columns”=“1,2,3,4,5”,
“replication_num” = “3”,
“dynamic_partition.enable” = “true”,
“dynamic_partition.time_unit” = “DAY”,
“dynamic_partition.start” = “-30”,
“dynamic_partition.end” = “3”,
“dynamic_partition.prefix” = “p”,
“dynamic_partition.buckets” = “48”
);

导入的配置：
这样的导入任务我起了4个的时候消费数据达到最大化。
后来加到8个也不会提升了。
CREATE ROUTINE LOAD load_00008 ON v6
COLUMNS (100列),
WHERE rule_group_id !=""
PROPERTIES
(
“format”=“json”,
“desired_concurrent_number”=“13”,
“max_error_number”=“200000”
)
FROM KAFKA
(
“kafka_broker_list”=“xxx”,
“kafka_topic”=“xx”,
“property.security.protocol”=“SASL_PLAINTEXT”,
“property.sasl.mechanism”=“GSSAPI”,
“property.sasl.kerberos.service.name”=“kafka”,
“property.sasl.kerberos.keytab”=“xxx”,
“property.sasl.kerberos.principal”=“xxxxx”,
“property.group.id” = “starrocks_v6_load”,
“kafka_partitions” = “105,106,107,108,109,110,111,112,113,114,115,116,117,118,119”,
“kafka_offsets” = “1162834805,1181955137,1111741807,1124902750,1255682692,1135177260,1217492156,1173053817,1323193605,1119963261,1142216733,1167423810,1248097662,1158699173,1273511598”
);

#3

fe的配置。我调试的时候改了改，改了的参数如下:
load_parallel_instance_num = 2
routine_load_task_consume_second = 60
routine_load_task_timeout_second = 600
alter_table_timeout_second = 432000
max_routine_load_task_concurrent_num = 15
max_routine_load_batch_size=2147483648
max_routine_load_task_num_per_be = 10

be同上:
push_write_mbytes_per_sec = 100
routine_load_thread_pool_size = 24
publish_version_worker_count = 2
max_consumer_num_per_group = 3
cumulative_compaction_num_threads_per_disk = 4
base_compaction_num_threads_per_disk = 2
cumulative_compaction_check_interval_seconds = 5
max_compaction_concurrency = 10
base_compaction_write_mbytes_per_sec=60
cumulative_compaction_write_mbytes_per_sec=300
alter_tablet_worker_count=8
memory_limitation_per_thread_for_schema_change = 6

#4

升级到2.1.8版本
修改下列参数，其他参数不用变，再起8个导入任务试一下。
BE：
txn_commit_rpc_timeout_ms=40000
quorom_publish_wait_time_ms=20000
routine_load_thread_pool_size=10

FE:
max_routine_load_batch_size=4294967295

另外任务消费的partition_num % (desired_concurrent_number * max_consumer_num_per_group)最好等于0，可以调整一下参数的组合。

#5

我升级到2.1.8版本并且更改了参数。但是看导入速率还是在400mb/s

我发现别的导入任务导入速率就挺快的，会不会是我加过滤条件导致导入变慢么？
WHERE rule_group_id !="" 。我看file_scan_node里会执行这个表达式。
数据源的数据，这个字段有时候是null的 有时候是""空字符串。这样会有影响吗？

#6

您好，当时解决了吗？

京ICP备2022026421号-1