点一下关注吧！！！非常感谢！！持续更新！！！
目前已经更新到了：
Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（正在更新！）
章节内容
上节我们完成了如下的内容：
Spark Streaming DStream 有状态转换
DStream 有状态转换 案例
基础介绍
针对不同的Spark、Kafka版本，集成处理数据的方式有两种：
- Receiver Approach
- Direct Approach
对应的版本：
版本的发展：
Kafka-08 接口
Receiver based Approach
基于 Receiver 的方式使用 Kafka 旧版本消费者高阶 API 实现。
对于所有的 Receiver，通过 Kafka 接收的数据被存储于 Spark 的 Executors 上，底层是写入 BlockManager中，默认200ms生成一个block（spark.streaming.blockInterval）
然后由SparkStreaming提交的Job构建BlockRDD，最终以SparkCore任务的形式运行。
对应Receiver方式，有以下几点需要注意：
Receiver 作为一个常驻线程调度到Executor上运行，占用一个CPU
Receiver 个数由 KafkaUtils.createStream 调用次数决定，一次一个Receiver
Kafka 中的Topic分区并不能关联产生在 SparkStreaming中的RDD分区，增加在KafkaUtils.createStream() 中的指定的Topic分区数，仅仅增加了单个Receiver消费的Topic的线程数，它不会增加处理数据中的并行的Spark的数量。
Receiver默认200ms生成一个Block，可根据数量大小调整Block生成周期，一个Block对应RDD一个分区
Receiver接收的数据会放入到BlockManager，每个Executor都会有一个BlockManager实例，由于数据本地性，那些存在 Receiver的Executor会被调度执行更多的Task，就会导致某些Executor比较空闲
默认情况下，Receiver是可能丢数据的，可以通过设置spark.streaming.receiver.writeAheadLog.enable为true开启预写日志机制，将数据先写入一个可靠的分布式文件系统（HDFS），确保数据不丢失，但会损失一定的性能
Kafka-08接口（Receiver方式）
- Offset 保存在ZK中，系统管理
- 对应Kafka版本 0.8.2.1 +
- 接口底层实现使用Kafka旧版消费者 高阶API
- DStream底层实现为BlockRDD
Kafka-08接口（Receiver with WAL）
- 增强了故障恢复的能力
- 接收的数据与Driver的元数据保存到HDFS
- 增加了流式应用处理的延迟
-
Direct Approach 是 Spark Streaming 不使用 Receiver 集成 Kafka 的方式，在企业生产环境中使用较多，相较于 Receiver，有以下特点：
不使用 Receiver，减少不必要的CPU占用，减少了 Receiver接收数据写入BlockManager，然后运行时再通过 BlockId、网络传输、磁盘读取等来获取数据的整个过程，提升了效率，无需WAL，进一步减少磁盘IO
Direct方式生的RDD是KafkaRDD，它的分区数与Kafka分区数保持一致，便于把控并行度。注意：在Shuffle 或 Repartition 操作后生成的 RDD，这种对应关系会失效
可以手动维护 Offset，实现 Exactly Once 语义
Kafka-10 接口
Spark Streaming 与 Kafka 0.10整合，和 0.8版本的Direct方式很像，Kafka的分区和Spark的RDD分区是一一对应的，可以获取 Offsets 和 元数据，API使用起来没有显著的区别。
添加依赖
<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-10_2.12</artifactId> <version>${spark.version}</version> </dependency>
不要手动添加 org.apache.kafka相关的依赖，如 kafka-clients，spark-streaming-kafka-0-10已经包含相关的依赖了，不同的版本会有不同程度的不兼容。
使用 kafka010接口从Kafka中获取数据：
Kafka集群
Kafka生产者发送数据
Spark Streaming 程序接收数
KafkaProducer
编写代码
package icu.wzk import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} import org.codehaus.jackson.map.ser.std.StringSerializer import java.util.Properties object KafkaProducerTest { def main(args: Array[String]): Unit = { // 定义 Kafka 参数 val brokers = "h121.wzk.icu:9092" val topic = "topic_test" val prop = new Properties() prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers) prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer]) prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer]) // KafkaProducer val producer = new KafkaProducer[String, String](prop) for (i <- 1 to 1000) { val msg = new ProducerRecord[String, String](topic, i.toString, i.toString) // 发送消息 producer.send(msg) println(s"i = $i") Thread.sleep(100) } producer.close() } }
运行测试
i = 493 i = 494 i = 495 i = 496 i = 497 i = 498 i = 499 i = 500 i = 501 i = 502 i = 503 i = 504
运行过程截图为：
查看Kafka
我们在服务器上查看当前Kafka中的队列信息：
kafka-topics.sh --list --zookeeper h121.wzk.icu:2181
可以看到队列已经加入了，spark_streaming_test01：
KafkaDStream
编写代码
package icu.wzk import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord} import org.apache.kafka.common.serialization.StringDeserializer import org.apache.log4j.{Level, Logger} import org.apache.spark.SparkConf import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies} import org.apache.spark.streaming.{Seconds, StreamingContext} object KafkaDStream1 { def main(args: Array[String]): Unit = { Logger.getLogger("org").setLevel(Level.ERROR) val conf = new SparkConf() .setAppName("KafkaDStream1") .setMaster("local[*]") val ssc = new StreamingContext(conf, Seconds(2)) val kafkaParams: Map[String, Object] = getKafkaConsumerParameters("wzkicu") val topics: Array[String] = Array("spark_streaming_test01") // 从 Kafka 中获取数据 val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils .createDirectStream( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)) // dstream输出 dstream.foreachRDD { (rdd, time) => if (!rdd.isEmpty()) { println(s"========== rdd.count = ${rdd.count()}, time = $time ============") } } ssc.start() ssc.awaitTermination() } private def getKafkaConsumerParameters(groupId: String): Map[String, Object] = { Map[String, Object]( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "h121.wzk.icu:9092", ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer], ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer], ConsumerConfig.GROUP_ID_CONFIG -> groupId, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest", ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean) ) } }
运行结果
WARNING: An illegal reflective access operation has occurred WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/Users/wuzikang/.m2/repository/org/apache/spark/spark-unsafe_2.12/2.4.5/spark-unsafe_2.12-2.4.5.jar) to method java.nio.Bits.unaligned() WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations WARNING: All illegal access operations will be denied in a future release Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties ========== rdd.count = 1000, time = 1721721502000 ms ============
运行截图如下:
生成数据
继续启动 KafkaProducer 的程序，让数据不断地写入
我们会看到控制台输出内容如下：
========== rdd.count = 1000, time = 1721721502000 ms ============ ========== rdd.count = 9, time = 1721721710000 ms ============ ========== rdd.count = 19, time = 1721721712000 ms ============ ========== rdd.count = 19, time = 1721721714000 ms ============ ========== rdd.count = 19, time = 1721721716000 ms ============ ========== rdd.count = 20, time = 1721721718000 ms ============ ========== rdd.count = 19, time = 1721721720000 ms ============ ========== rdd.count = 19, time = 1721721722000 ms ============ ========== rdd.count = 19, time = 1721721724000 ms ============
运行结果如下图所示：
