实验五 Spark Streaming编程初级实践

06-30 1535阅读

1 实验目的

（1）通过实验学习使用Scala编程实现文件和数据的生成。

（2）掌握使用文件作为Spark Streaming数据源的编程方法。

2 实验平台

操作系统：Ubuntu16.04及以上。

Spark版本：3.4.0。

Scala版本：2.12.17。

3 实验要求

产生一系列字符串的程序，会产生随机的整数序列，每个整数被当做一个单词，提供给KafkaWordCount程序去进行词频统计

4 实验内容和步骤（操作结果要附图）

Kafka准备工作（执行如下命令完成Kafka的安装：）

这里使用的软件版本是：kafka_2.12-2.6.0，Spark3.4.0（Scala版本是2.12.17）

链接: https://pan.baidu.com/s/1RS42IvXxtTVvIL24IwcjTA?pwd=1234 提取码: 1234

安装kafka

cd ~/Downloads
sudo tar -zxf  kafka_2.12-2.6.0.tgz -C /usr/local
cd /usr/local
sudo mv kafka_2.12-2.6.0 kafka
sudo chown -R hadoop ./kafka

启动Kafka

第一个终端

cd /usr/local/kafka
./bin/zookeeper-server-start.sh config/zookeeper.properties

行上面命令以后，终端窗口会返回一堆信息，然后就停住不动了，是Zookeeper服务器启动了，正在处于服务状态。所以，千万不要关闭这个终端窗口，一旦关闭，zookeeper服务就停止了，所以，不能关闭这个终端窗口

再打开第二个（不能关闭）

cd /usr/local/kafka
bin/kafka-server-start.sh config/server.properties

再打开第三个终端，然后输入下面命令创建一个自定义名称为“wordsender”的Topic：

cd /usr/local/kafka
./bin/kafka-topics.sh --create --zookeeper localhost:2181 \
> --replication-factor 1 --partitions 1 \
> --topic wordsender

查看名称为“wordsender”的Topic是否已经成功创建：

./bin/kafka-topics.sh --list --zookeeper localhost:2181

再新开一个终端（记作“监控输入终端”），执行如下命令监控Kafka收到的文本(可以尝试输入一下，再另一个窗口可以收到)：

cd /usr/local/kafka
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic wordsendertest --from-beginning

cd /usr/local/spark
./bin/spark-shell

注意，所有这些终端窗口都不要关闭，要继续留着后面使用。

5 Spark添加相关jar包

下载spark-streaming-kafka-0-10_2.12-3.4.0.jar和spark-token-provider-kafka-0-10_2.12-3.4.0.jar文件，其中，2.12表示Scala的版本号，3.4.0表示Spark版本号。然后，把这两个文件复制到Spark目录的jars目录下（即“/usr/local/spark/jars”目录）。此外，还需要把“/usr/local/kafka/libs”目录下的kafka-clients-2.6.0.jar文件复制到Spark目录的jars目录下。（链接在上面）

import org.apache.spark.streaming.kafka010._

（在kafka0-10版本之后，接口发生了变化，在原有基础上加入010

参考文献：Spark Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher) - Spark 3.5.1 Documentation (apache.org)https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html）

编写Spark程序使用Kafka数据源（又一个新终端）


cd  /usr/local/spark/mycode
mkdir  kafka
cd  kafka
mkdir  -p  src/main/scala
cd  src/main/scala

import java.util.HashMap
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
object KafkaWordProducer {
  def main(args: Array[String]) {
    if (args.length 
        val str = (1 to wordsPerMessage.toInt).map(x => scala.util.Random.nextInt(10).
toString)
          .mkString(" ")
                    print(str)
                    println()
        val message = new ProducerRecord[String, String](topic, null, str)
        producer.send(message)
      }
     Thread.sleep(1000)
    }
  }
}

vim KafkaWordCount.scala

KafkaWordCount.scala是用于单词词频统计，它会把KafkaWordProducer发送过来的单词进行词频统计，代码内容如下：

import org.apache.spark._
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
object KafkaWordCount{
  def main(args:Array[String]){
    val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc,Seconds(10))
    ssc.checkpoint("file:///usr/local/spark/mycode/kafka/checkpoint") //设置检查点，如果存放在HDFS上
面，则写成类似ssc.checkpoint("/user/hadoop/checkpoint")这种形式，但是，要启动Hadoop
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "use_a_separate_group_id_for_each_stream",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean)
    )
    val topics = Array("wordsender")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    stream.foreachRDD(rdd => {
      val offsetRange = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      val maped: RDD[(String, String)] = rdd.map(record => (record.key,record.value))
      val lines = maped.map(_._2)
      val words = lines.flatMap(_.split(" "))
      val pair = words.map(x => (x,1))
      val wordCounts = pair.reduceByKey(_+_)
      wordCounts.foreach(println)
    })
    ssc.start
    ssc.awaitTermination
  }
}

下面是StreamingExamples.scala的代码，用于设置log4j:

vim StreamingExamples.scala

package org.apache.spark.examples.streaming
import org.apache.spark.internal.Logging
import org.apache.log4j.{Level, Logger}
object StreamingExamples extends Logging {
  def setStreamingLogLevels(): Unit = {
    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if (!log4jInitialized) {
      logInfo("Setting log level to [WARN] for streaming example. " +
              "To override add a custom log4j.properties to the classpath.")
      Logger.getRootLogger.setLevel(Level.WARN)
    }
  }
}

三个文件已经创建好了

cd /usr/local/spark/mycode/kafka/
vim simple.sbt

name := "Simple Project"
version := "1.0"
scalaVersion := "2.12.17"
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.4.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "3.4.0" % "provided"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "3.4.0"
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "2.6.0"

然后执行下面命令，进行编译打包：

cd  /usr/local/spark/mycode/kafka/
/usr/local/sbt/sbt  package

首先，启动Hadoop，因为如果前面KafkaWordCount.scala代码文件中采用了ssc.checkpoint
("/user/hadoop/checkpoint")这种形式，这时的检查点是被写入HDFS，因此需要启动Hadoop。启动Hadoop的命令如下：

cd  /usr/local/hadoop
./sbin/start-dfs.sh

要注意，之前已经启动了Zookeeper服务和Kafka服务，因为之前那些终端窗口都没有关闭，所以，这些服务一直都在运行。如果不小心关闭了之前的终端窗口，那就参照前面的内容，再次启动Zookeeper服务，启动Kafka服务。
然后，新打开一个终端，执行如下命令，运行“KafkaWordProducer”程序，生成一些单词（是一堆整数形式的单词）：

cd  /usr/local/spark/mycode/kafka/
/usr/local/spark/bin/spark-submit  \
> --class "KafkaWordProducer"   \
> ./target/scala-2.12/simple-project_2.12-1.0.jar  \
> localhost:9092  wordsender  3  5

注意，上面命令中，“localhost:9092 wordsender 3 5”是提供给KafkaWordProducer程序的4个输入参数，第1个参数“localhost:9092”是Kafka的Broker的地址，第2个参数“wordsender”是Topic的名称，我们在KafkaWordCount.scala代码中已经把Topic名称写死掉，所以，KafkaWordCount程序只能接收名称为“wordsender”的Topic。第3个参数“3”表示每秒发送3条消息，第4个参数“5”表示每条消息包含5个单词（实际上就是5个整数）。

不要关闭这个终端窗口，让它一直不断发送单词。然后，再打开一个终端，执行下面命令，运行KafkaWordCount程序，执行词频统计：

cd  /usr/local/spark/mycode/kafka/
/usr/local/spark/bin/spark-submit  \
> --class "KafkaWordCount"  \
> ./target/scala-2.12/simple-project_2.12-1.0.jar

屏幕上就会显示如下类似信息

5 实验总结

确保在 Spark 环境中配置了正确的 Kafka 相关依赖，包括 Kafka 客户端和 Spark Streaming 对 Kafka 的集成依赖。确保 Kafka 集群正常运行，并且你有权限访问 Kafka 集群。
在 Spark Streaming 应用中配置 Kafka 数据源，包括指定 Kafka 主题、Kafka 集群地址、消费者组等信息。

编写 Spark Streaming 应用程序，包括创建 SparkContext 和 StreamingContext，以及定义数据处理逻辑。

3.总的来说，运行 Spark Streaming 与 Kafka 数据源的实验需要综合考虑环境配置、数据处理逻辑、调试与监控等方面，并不断优化和完善应用程序，以满足实际业务需求并保证系统的稳定性和可靠性。

VPS购买请点击我