spark-streaming
DStream创建Kafka数据源：
ReceiverAPI：需要一个专门的 Executor 去接收数据，然后发送给其他的 Executor 做计算。存在的问题，接收数据的 Executor 和计算的 Executor 速度会有所不同，特别在接收数据的 Executor速度大于计算的 Executor 速度，会导致计算数据的节点内存溢出。
DirectAPI：是由计算的 Executor 来自动消耗 Kafka 的数据，速度由自身控制。
Kafka 0-10 Direct 模式
需求：通过 SparkStreaming 从 Kafka 读取数据，并将读取过来的数据做简朴计算，最终打印到控制台。
导入依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
编写代码
/**
* 通过DirectAPI 0-10 消耗kafka数据
* 消耗的offset生存在_consumer_offsets主题中
*/
object DirectAPI {
def main(args: Array): Unit = {
val sparkConf = new SparkConf().setMaster("local
[*]").setAppName("direct")
val ssc = new StreamingContext(sparkConf,Seconds(3))
//界说kafka相关参数
val kafkaPara :Map = Map(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->"node01:9092,node02:9092,node03:9092",
ConsumerConfig.GROUP_ID_CONFIG->"kafka",
"key.deserializer"->"org.apache.kafka.common.serialization.StringDeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
//通过读取kafka数据，创建DStream
val kafkaDStream:InputDStream] = KafkaUtils.createDirectStream(
ssc,LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe(Set("kafka"),kafkaPara)
)
//提取出数据中的value部分
val valueDStream :DStream = kafkaDStream.map(record=>record.value())
//wordCount计算逻辑
valueDStream.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.print()
ssc.start()
ssc.awaitTermination()
开启Kafka集群
https://i-blog.csdnimg.cn/direct/76a13a942f464045a2ee675961e142cc.png
开启Kafka生产者，产生数据
kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic kafka
https://i-blog.csdnimg.cn/direct/1b4034b5fa3349b69e5592b13eebbf0d.png
运行程序，接收Kafka生产的数据并举行相应处置惩罚
https://i-blog.csdnimg.cn/direct/60d40f8da61d4b16ac813cf84f1f24dd.png
查看消耗进度
kafka-consumer-groups.sh --describe --bootstrap-server node01:9092,node02:9092,node03:9092 --group kafka
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]