本文基于Spark2.1.0版本
雖然很少有生產(chǎn)環(huán)境用Kafka傳遞超過(guò)1M消息的場(chǎng)景（因?yàn)楦咄掏?、低延時(shí)的要求，Kafka 發(fā)布-訂閱模型中Producer-Broker-Consumer 3方的相關(guān)默認(rèn)配置都是1M），但由于手上項(xiàng)目的特殊需求，希望Spark Streaming抽取Kafka數(shù)據(jù)源時(shí)，能消費(fèi)30M-40MB大小的消息。下面我把相關(guān)配置及源碼提供一下，有需要的同學(xué)可以參考。
（本文只涉及ETL過(guò)程中數(shù)據(jù)的抽取，不涉及轉(zhuǎn)換和加載過(guò)程）
業(yè)務(wù)模型如下：
1，修改Producer-Broker-Consumer 3方配置，使其支持超大消息的傳遞和接收
A：修改Producer的max.request.size，允許生產(chǎn)者發(fā)送超大消息，默認(rèn)是1M
vim $KAFKA_HOME/config/producer.properties
修改max.request.size=41943040 ?#40MB
修改Producer的buffer.memory，允許生產(chǎn)者發(fā)送超大消息時(shí)的緩沖區(qū)，默認(rèn)是32M
buffer.memory=45000000 # >40MB
B：修改Broker的message.max.bytes、replica.fetch.max.bytes、max.message.bytes允許Broker傳遞和備份超大消息，默認(rèn)是1M
vim $KAFKA_HOME/config/server.properties
修改message.max.bytes=41943040? #40MB
修改replica.fetch.max.bytes=41943040? #40MB
max.message.bytes可以不用修改，默認(rèn)值=message.max.bytes
C：修改Consumer的fetch.message.max.bytes，允許消費(fèi)者拉取超大消息，默認(rèn)是1M
vim $KAFKA_HOME/config/consumer.properties
修改fetch.message.max.bytes=41943040? #40MB
上述配置修改完成后，所有Broker都需要重啟Kafka服務(wù)。
2，Spark Streaming用Kafka 0.10.2的new Kafka consumer API 來(lái)消費(fèi)超大消息
3，測(cè)試
通過(guò)$KAFKA_HOME/bin下的kafka-producer-perf-test.sh充當(dāng)生產(chǎn)者，發(fā)送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
生產(chǎn)者發(fā)送消息后，通過(guò)Kafka Manager觀察，Consumer已經(jīng)完成了消息抽取后的commit，Consumer offset增加。
driver端輸出消息
web ui能看到消費(fèi)的具體情況
4，Spark 2.1.0 的Streaming使用Kafka 0.10.2新Consumer API的整合接口好處多多
詳見官網(wǎng)：
spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
