大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka(已更完)

Spark(正在更新!)

章节内容

上节我们完成了如下的内容:


Spark Streaming DStream 有状态转换

DStream 有状态转换 案例

64e7805c52806f8d3f7fe684f7039c95_c83c783141c147b9a528f709377b64c6.png

基础介绍

针对不同的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方式)

c9c8f15b5195cc90149833797593913e_ba01e98a1e074e06bba5f2ddc4033455.png

  • 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 ============

运行结果如下图所示:

目录
相关文章
|
3月前
|
人工智能 分布式计算 大数据
大数据≠大样本:基于Spark的特征降维实战(提升10倍训练效率)
本文探讨了大数据场景下降维的核心问题与解决方案,重点分析了“维度灾难”对模型性能的影响及特征冗余的陷阱。通过数学证明与实际案例,揭示高维空间中样本稀疏性问题,并提出基于Spark的分布式降维技术选型与优化策略。文章详细展示了PCA在亿级用户画像中的应用,包括数据准备、核心实现与效果评估,同时深入探讨了协方差矩阵计算与特征值分解的并行优化方法。此外,还介绍了动态维度调整、非线性特征处理及降维与其他AI技术的协同效应,为生产环境提供了最佳实践指南。最终总结出降维的本质与工程实践原则,展望未来发展方向。
195 0
|
6月前
|
存储 分布式计算 Hadoop
从“笨重大象”到“敏捷火花”:Hadoop与Spark的大数据技术进化之路
从“笨重大象”到“敏捷火花”:Hadoop与Spark的大数据技术进化之路
287 79
|
10月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
679 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
10月前
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第27天】在大数据时代,数据湖技术凭借其灵活性和成本效益成为企业存储和分析大规模异构数据的首选。Hadoop和Spark作为数据湖技术的核心组件,通过HDFS存储数据和Spark进行高效计算,实现了数据处理的优化。本文探讨了Hadoop与Spark的最佳实践,包括数据存储、处理、安全和可视化等方面,展示了它们在实际应用中的协同效应。
445 2
|
10月前
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第26天】本文详细探讨了Hadoop与Spark在大数据处理中的协同作用,通过具体案例展示了两者的最佳实践。Hadoop的HDFS和MapReduce负责数据存储和预处理,确保高可靠性和容错性;Spark则凭借其高性能和丰富的API,进行深度分析和机器学习,实现高效的批处理和实时处理。
393 1
|
消息中间件 分布式计算 Kafka
195 Spark Streaming整合Kafka完成网站点击流实时统计
195 Spark Streaming整合Kafka完成网站点击流实时统计
151 0
|
11月前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
168 0
|
11月前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
215 0
|
消息中间件 分布式计算 Kafka
大数据Spark Structured Streaming集成 Kafka
大数据Spark Structured Streaming集成 Kafka
281 0
|
消息中间件 分布式计算 Kafka
大数据Spark Streaming集成Kafka
大数据Spark Streaming集成Kafka
294 0

热门文章

最新文章