大数据学习笔记-------------------(18)

大数据学习笔记-------------------(18)

第18章 KAFKA与SPARK集成

18.1Kafka与spark集成

Kafka是Spark流式传输潜在消息和集成的平台。Kafka充当实时数据流的中心枢纽，并使用Spark Streaming中的复杂算法进行处理。一旦数据被处理，Spark Streaming可以将结果发布到另一个Kafka topic或存储在HDFS，数据库或控制面板中。

18.2 SparkConfAPI

SparkConf表示Spark应用程序的配置。用于将各种Spark参数设置为键值对。SparkConf类有如下方法：

Ø  set(string key,stringvalue)----------设置配置变量

Ø  remove(string key)--------------------从配置中移除key

Ø  setAppName(string name)---------设置应用程序名称

Ø  get(string key)--------------------------获取key

18.3StreamingContext API

Spark功能的主要入口点。SparkContext表示到Spark集群的连接，可用于在集群上创建RDD，累加器和广播变量。签名的定义如下所示:

public StreamingContext(String master, String appName, Duration batchDuration, String sparkHome, scala.collection.Seq jars, scala.collection.Map environment)

master：连接打集群上的URL

appName：作业名称，并显示在集群WebUI上

batchDuration：定期将流数据分进批次

public StreamingContext(SparkConf conf, Duration batchDuration)

conf：Spark 参数

batchDuration：定期将流数据分进批次​

18.4 KafkaUtilsAPI

KafkaUtils API用于将Kafka集群连接到Spark流。API具有如下定义方法createStream签名：

public static ReceiverInputDStream> createStream( StreamingContext ssc, String zkQuorum, String groupId, scala.collection.immutable.Map topics, StorageLevel storageLevel)

ssc：StreamingContext 对象

zkQuorum：zookeeper法定人数

groupId：消费者group id

topics：返回用于消费topics 的map

storageLever：Storage level用于存储接收的对象

KafkaUtilsAPI有另一个方法createDirectStream，用于创建一个输入流，直接从Kafka Brokers拉取消息，而不使用任何接收器。这个流可以保证来自一次转换中Kafka的每个消息。

示例应用程序。 要编译应用程序，请下载并安装“sbt”，scala构建工具（类似于maven）。主要应用程序代码如下所示：

import java.util.HashMapimport org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}import org.apache.spark.SparkConfimport org.apache.spark.streaming._import org.apache.spark.streaming.kafka._object KafkaWordCount { def main(args: Array[String]) { if (args.length < 4) { System.err.println("Usage: KafkaWordCount ") System.exit(1) } val Array(zkQuorum, group, topics, numThreads) = args val sparkConf = new SparkConf().setAppName("KafkaWordCount") val ssc = new StreamingContext(sparkConf, Seconds(2)) ssc.checkpoint("checkpoint") val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2) val words = lines.flatMap(_.split(" ")) val wordCounts = words.map(x => (x, 1L)) .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2) wordCounts.print() ssc.start() ssc.awaitTermination() }}

18.5 建立脚本

spark-kafka集成取决于spark，spark流和spark Kafka集成jar。 创建一个新文件“build.sbt”并指定应用程序详细信息及其依赖关系。 “sbt”将在编译和打包应用程序时下载必要的jar:

name := "Spark Kafka Project"version := "1.0"scalaVersion := "2.10.5"libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"

18.6 编译/打包

运行以下命令以编译和打包应用程序的jar文件。将jar文件提交到spark控制台以运行应用程序：sbt package

18.7 提交到Spark

启动kafka 生产者CLI，创建一个新的topic叫"myfirst-topic"提供一些例子信息如下：

Anotherspark test message

在spark 控制台运行如下命令提交应用程序：

/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming-kafka_2.10:1.6.0 --class "KafkaWordCount" --master local[4] target/scala-2.10/spark-kafka-project_2.10-1.0.jar localhost:2181

应用程序输出：

spark console messages ..(Test,1)(spark,1)(another,1)(message,1)spark console message ..

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。