本文基于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的整合接口好处多多
详见官网：
spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
网友评论