点燃激情，踏上Spark与Kafka的征程：SparkStreaming Direct方式构建实时数据应用

后端

2023-08-22 07:31:15

用 Spark Streaming 和 Kafka 踏上大数据之旅

初探 Spark Streaming

Spark Streaming，一个强大的分布式实时数据处理框架，让你轻松构建实时数据应用。它能从 Kafka、Flume 和 Twitter 等多种数据源读取数据，并支持一系列数据处理操作，如过滤、聚合和连接。

理解 Kafka

Kafka，一个分布式消息系统，可让你轻松发布和消费数据。其高吞吐量、低延迟和可靠性使其成为实时数据处理的理想选择。

配置 Kafka 集群

首先，你需要配置一个 Kafka 集群。推荐使用 Confluent Platform，一个经过优化的 Kafka 发行版，可轻松部署和管理集群。

启动 Kafka 集群

配置好集群后，启动它。使用命令行工具或 Confluent Control Center 均可。

创建 Kafka 主题

主题是 Kafka 中存储数据的逻辑单元。使用命令行工具或 Confluent Control Center 创建主题。

运行 Scala 消费者程序

用 Scala 开发消费者程序，它从订阅的主题消费数据并处理。

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf

// Spark 配置
val conf = new SparkConf().setAppName("SparkStreamingKafkaDirectWordCount")

// 流上下文，批处理间隔 5 秒
val ssc = new StreamingContext(conf, Seconds(5))

// Kafka 参数
val kafkaParams = Map[String, String](
  "bootstrap.servers" -> "localhost:9092",
  "group.id" -> "spark-streaming-kafka-direct-wordcount",
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)

// 订阅主题
val topics = Array("my-topic")

// 创建 Kafka 直连流
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

// 拆分行并统计词频
val wordCounts = stream.map(_.value).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)

// 打印词频
wordCounts.print()

// 启动流上下文
ssc.start()

// 等待终止
ssc.awaitTermination()