一、实验目的

掌握Kafka作为Spark结构化流的外部数据源。

  掌握kafka作为Spark的Sink。

二、实验内容

1、在Spark结构化流中连接Kafka，将Kafka作为外部数据源。

  2、将Spark结构化流程序计算后的结果写入Kafka。

三、实验原理

Kafka通常用于构建实时流数据管道，以可靠地在系统之间移动数据，还用于转换和响应数据流。Kafka作为集群在一个或多个服务器上运行。

 Kafka的一些关键概念在这里描述：

• Topic：消息发布到的类别或流名称的高级抽象。主题可以有0、1或多个消费者，这些消费者订阅发布到该主题的消息。用户为每个新的消息类别定义一个新主题；

• Producers：向主题发布消息的客户端；

• Consumers：使用来自主题的消息的客户端；

• Brokers：复制和持久化消息数据的一个或多个服务器。

  默认情况下，Kafka的数据源并不是Spark的内置数据源。如果我们要从spark shell使用Kafka数据源，那么需要在启动spark shell时将依赖的JAR包添加到classpath中。可手动将依赖包添加到classpath。

四、实验环境

硬件：x86_64 ubuntu 16.04服务器

  软件：JDK 1.8，Spark-2.3.2，Hadoop-2.7.3

五、实验步骤

以下操作均在”VNC界面”下操作。请首先切换到VNC界面。

5.1 启动zookeeper服务和kafka服务

1、启动zookeeper服务

  Kafka依赖于Apache ZooKeeper，所以在启动Kafka之前，要先启动它。打开一个终端窗口，执行如下命令：

1.  $ cd /opt/kafka
2.  $ ./bin/zookeeper-server-start.sh config/zookeeper.properties

这将在2181端口启动ZooKeeper进程

  2、启动kafaka服务

  接下来，启动Kafka服务器。另打开一个终端窗口，执行如下命令：

1.  $ cd /opt/kafka
2.  $ ./bin/kafka-server-start.sh config/server.properties

3、创建用于发送数据的主题”topic1”和接收计算结果数据的主题”topic2”。另外打开第三个终端窗口，执行如下命令：

1.  $ cd /opt/kafka
2.  $ ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic1
3.  $ ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic2

4、浏览所创建的主题。继续在第三个终端窗口下，执行如下命令：

1.  $ ./bin/kafka-topics.sh --list --bootstrap-server localhost:9092

应该可以看到刚才创建的主题”topic1”和”topic2”。

5.2 启动HDFS和Spark集群

在终端窗口下，输入以下命令，启动HDFS和Spark集群：

1.  $ start-dfs.sh
2.  $ cd /opt/spark
3.  $ ./sbin/start-all.sh

5.3 编写Spark结构化流程序

按下来，编写Spark结构化流程序，读取Kafa的”topic1”主题中的内容，经过简单处理，将计算结果写出到Kafka的”topic2”主题。请按以下步骤操作：

1、启动spark-shell。在终端窗口下，执行如下命令：(注意，请将命令中的localhost替换为当前实际的机器名)

1.  $ spark-shell --master spark://localhost:7077 --jars /data/software/spark-sql-kafka-0-10_2.11-2.3.2.jar,/data/software/kafka-clients-2.4.1.jar

2、读取Kafa的”topic1”主题中的内容。在spark-shell窗口，输入如下代码:

1.  val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "topic1").load()

3、将消息内容转换为String类型，再从DataFrame转为DataSet[String]。在spark-shell窗口，输入如下代码:

1.  import spark.implicits._
2.  val content = df.selectExpr("CAST(value AS STRING)").as[String]

4、对从”topic1”抓取的消息进行转换。在spark-shell窗口，输入如下代码:

1.  import org.apache.spark.sql.functions.upper
2.  val resultDf = content.withColumn("value", upper(col("value")));

5、将转换后的DataFrame再写入到Kafka的”topic2”主题中。在spark-shell窗口，进入到paste模式，然后输入如下代码:

1.  val sinkQuery = resultDf.writeStream
2.                          .format("kafka")
3.                          .outputMode("append")
4.                          .option("kafka.bootstrap.servers", "localhost:9092")
5.                          .option("topic","topic2")
6.                          .option("checkpointLocation", "/ck/streaming_kafka")
7.                          .start()

5.4 执行流处理程序

请按以下步骤，执行流处理程序。

  1. 另外再打开一个终端窗口（第4个终端窗口），在这个终端窗口中，启动Kakfa自带的生产者客户端脚本程序，向Kafka的”topic1”主题写入一些消息。在终端窗口下，键入如下命令：

1.  $ cd /opt/kafka
2.  $ ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1

2. 另外再打开一个终端窗口（第5个终端窗口），在这个终端窗口中，启动Kakfa自带的消费者客户端脚本程序，它会消费来自”topic2”主题的消息，查看我们的流程序的输出内容。在终端窗口下，键入如下命令：

1.  $ cd /opt/kafka
2.  $ ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic2

3. 切换到第4个终端窗口，输入以下内容，这些内容会被写入到Kafka的”topic1”主题，里面被我们的Spark流程序实时处理，并把处理结果写出到Kafka的”topic2”主题中：

1.  hello spark

4. 切换到第5个终端窗口，会看到如下所示的输出：

HELLO SPARK

5.等待所有的文件都被处理完后，从shell停止运算流。切换到第3个终端窗口，执行以下代码：

1.  sinkQuery.stop()

— END —