手记

Spark Streaming如何消费Kafka的大消息(30M-40MB)

本文基于Spark2.1.0版本

虽然很少有生产环境用Kafka传递超过1M消息的场景(因为高吞吐、低延时的要求,Kafka 发布-订阅模型中Producer-Broker-Consumer 3方的相关默认配置都是1M),但由于手上项目的特殊需求,希望Spark Streaming抽取Kafka数据源时,能消费30M-40MB大小的消息。下面我把相关配置及源码提供一下,有需要的同学可以参考。

(本文只涉及ETL过程中数据的抽取,不涉及转换和加载过程)

业务模型如下:

1,修改Producer-Broker-Consumer 3方配置,使其支持超大消息的传递和接收

producer配置

producer配置

A:修改Producer的max.request.size,允许生产者发送超大消息,默认是1M

vim $KAFKA_HOME/config/producer.properties

修改max.request.size=41943040  #40MB

修改Producer的buffer.memory,允许生产者发送超大消息时的缓冲区,默认是32M

buffer.memory=45000000 # >40MB

Broker配置

Broker配置

Broker配置

B:修改Broker的message.max.bytes、replica.fetch.max.bytes、max.message.bytes允许Broker传递和备份超大消息,默认是1M

vim $KAFKA_HOME/config/server.properties

修改message.max.bytes=41943040  #40MB

修改replica.fetch.max.bytes=41943040  #40MB

max.message.bytes可以不用修改,默认值=message.max.bytes

Consumer配置

C:修改Consumer的fetch.message.max.bytes,允许消费者拉取超大消息,默认是1M

vim $KAFKA_HOME/config/consumer.properties

修改fetch.message.max.bytes=41943040  #40MB

上述配置修改完成后,所有Broker都需要重启Kafka服务。

2,Spark Streaming用Kafka 0.10.2的new Kafka consumer API 来消费超大消息

使用spark-streaming-kafka-0-10_2.11-2.1.0.jar提供的KafkaUtils.createDirectStream接口

3,测试

通过$KAFKA_HOME/bin下的kafka-producer-perf-test.sh充当生产者,发送40MB左右的消息

./kafka-producer-perf-test.sh --topic wl_test --num-records 1 --record-size 40000000 --throughput  1 --producer-props bootstrap.servers=wl1:9092 max.request.size=45000000 buffer.memory=45000000


生产者发送消息后,通过Kafka Manager观察,Consumer已经完成了消息抽取后的commit,Consumer offset增加。

driver端输出消息

web ui能看到消费的具体情况

4,Spark 2.1.0 的Streaming使用Kafka 0.10.2新Consumer API的整合接口好处多多



作者:俺是亮哥
链接:https://www.jianshu.com/p/f44afb7c8709


0人推荐
随时随地看视频
慕课网APP