flink sink kafka过慢问题

[xiayongquan]

hi，all


我在使用flink处理数据写入kafka的过程中，写入kafka的速度过慢，导致数据高峰期数据有堆积，
在数据达到每秒11000条的时候就开始堆积，处理逻辑很简单，就有坐标点经纬度转换成省市区，其他逻辑不复杂，
sink端kafka 6个分区，source端kafka6个分区，每天在晚高峰的时候会堆积30分钟，有没有办法大幅度提高写入能力，
难道只能加大kafka的分区吗？


flink submit 参数如下
p=6（与kafka分区对应）
m=yarn-cluster
yjm=1024
ytm=4096
ys=6
checkpoint_timeout=30000
max_concurrent_checkpoint=1
checkpoint_between_time=1000
checkpoint_interval_ms=60000
restart_between_time=1000
restart_num=4
sink_buffer_memory=134217728（也尝试了很多）
sink_batch_size=65536（也尝试了很多）
sink_kafka_linger_ms=10（尝试过1 20 50 100 200 500）
sink_max_request_size=10240000
sink_kafka_acks=1


这个问题困扰我很久了，请大家帮忙看看 ，非常感谢！

[Leonard Xu]

Hi
看起来你经确定了是在写入kafka时过慢导致了高峰期sink反压，
生产环境上1万的tps不算高的，可以先确定下是否是kafka集群写入瓶颈问题，kafka 自带的 perf 脚本可以测试下6个分区的topic写入的能力，测试时数据大小和tpoic配置可以和生产一致，如果是kafka写入瓶颈问题，那就需要增加分区，对应修改flink作业的写入并发。
另外，如果开启exactly-once配置，写入速度会慢一些，没有特别的业务需求可以用at-least-once.

祝好
Leonard

> 在 2020年10月13日，19:38，xyq <[hidden email]> 写道：
>
> kafka

[hailongwang]

Hi xyq，
1. 可以确认下下游 kakfa 6个分区写入数据量都是均匀的吗，看下 Partitioner 有没有设置好。
2. 还有 11000 条的数据量大小有多少呢，有没有存在 flink 集群 与 kafka 集群 跨机房的限制。（在我们内部多个机房，其中延迟比较大的机房的速率只有 3M/s 单并发）
3. 可以确认下，Kafka sink 有没有一些耗时的序列化操作的
个人认为，这种问题我们最好分而治之，可以测试下 正常情况下两集群间的传输速率（比如scp），然后再看看 kafka 集群，然后 flink 任务等。
希望可以帮助到你


Best，
Hailong Wang




在 2020-10-13 18:38:37，"xyq" <[hidden email]> 写道：

>hi，all
>
>
>我在使用flink处理数据写入kafka的过程中，写入kafka的速度过慢，导致数据高峰期数据有堆积，
>在数据达到每秒11000条的时候就开始堆积，处理逻辑很简单，就有坐标点经纬度转换成省市区，其他逻辑不复杂，
>sink端kafka 6个分区，source端kafka6个分区，每天在晚高峰的时候会堆积30分钟，有没有办法大幅度提高写入能力，
>难道只能加大kafka的分区吗？
>
>
>flink submit 参数如下
>p=6（与kafka分区对应）
>m=yarn-cluster
>yjm=1024
>ytm=4096
>ys=6
>checkpoint_timeout=30000
>max_concurrent_checkpoint=1
>checkpoint_between_time=1000
>checkpoint_interval_ms=60000
>restart_between_time=1000
>restart_num=4
>sink_buffer_memory=134217728（也尝试了很多）
>sink_batch_size=65536（也尝试了很多）
>sink_kafka_linger_ms=10（尝试过1 20 50 100 200 500）
>sink_max_request_size=10240000
>sink_kafka_acks=1
>
>
>这个问题困扰我很久了，请大家帮忙看看 ，非常感谢！

[Xiao Xu]

建议先确认下瓶颈是不是 kafka sink, 一般来说 kafka 网卡打满都不会到瓶颈的, 猜测有可能其他逻辑导致的瓶颈

hailongwang <[hidden email]> 于2020年10月13日周二 下午10:22写道：

>
>
> Hi xyq，
> 1. 可以确认下下游 kakfa 6个分区写入数据量都是均匀的吗，看下 Partitioner 有没有设置好。
> 2. 还有 11000 条的数据量大小有多少呢，有没有存在 flink 集群 与 kafka 集群
> 跨机房的限制。（在我们内部多个机房，其中延迟比较大的机房的速率只有 3M/s 单并发）
> 3. 可以确认下，Kafka sink 有没有一些耗时的序列化操作的
> 个人认为，这种问题我们最好分而治之，可以测试下 正常情况下两集群间的传输速率（比如scp），然后再看看 kafka 集群，然后 flink 任务等。
> 希望可以帮助到你
>
>
> Best，
> Hailong Wang
>
>
>
>
> 在 2020-10-13 18:38:37，"xyq" <[hidden email]> 写道：
> >hi，all
> >
> >
> >我在使用flink处理数据写入kafka的过程中，写入kafka的速度过慢，导致数据高峰期数据有堆积，
> >在数据达到每秒11000条的时候就开始堆积，处理逻辑很简单，就有坐标点经纬度转换成省市区，其他逻辑不复杂，
> >sink端kafka 6个分区，source端kafka6个分区，每天在晚高峰的时候会堆积30分钟，有没有办法大幅度提高写入能力，
> >难道只能加大kafka的分区吗？
> >
> >
> >flink submit 参数如下
> >p=6（与kafka分区对应）
> >m=yarn-cluster
> >yjm=1024
> >ytm=4096
> >ys=6
> >checkpoint_timeout=30000
> >max_concurrent_checkpoint=1
> >checkpoint_between_time=1000
> >checkpoint_interval_ms=60000
> >restart_between_time=1000
> >restart_num=4
> >sink_buffer_memory=134217728（也尝试了很多）
> >sink_batch_size=65536（也尝试了很多）
> >sink_kafka_linger_ms=10（尝试过1 20 50 100 200 500）
> >sink_max_request_size=10240000
> >sink_kafka_acks=1
> >
> >
> >这个问题困扰我很久了，请大家帮忙看看 ，非常感谢！
>