Spark Streaming + Kafka整合

参考官网 http://spark.apache.org/docs/2.1.0/streaming-kafka-0-8-integration.html

  • 之前先确保以下操作: 1、先启动ZK:./zkServer.sh start 2、启动Kafka:./kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties 3、创建topic: ./kafka-topics.sh --create --zookeeper hadoop:2181 --replication-factor 1 --partitions 1 --topic kafka_streaming_topic ./kafka-topics.sh --list --zookeeper hadoop:2181 4、通过控制台测试是否能正常生产与消费 ./kafka-console-producer.sh --broker-list hadoop:9092 --topic kafka_streaming_topic ./kafka-console-consumer.sh --zookeeper hadoop:2181 --topic kafka_streaming_topic

Approach 1: Receiver-based Approach

  • Receiver方式的本地环境联调 1、KafkaUtils.createStream Create an input stream that pulls messages from Kafka Brokers.
import org.apache.spark.streaming.kafka._

 val kafkaStream = KafkaUtils.createStream(streamingContext,
     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

2、引入数组,含四个数->val Array(zkQuorum,group,topics,numThreads) = args

3、判断是否传入四个参数->构建topicMap: val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

4、topicMap带入KafkaUtils参数 5、业务代码: messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

6、到IDEA的edit configuration编辑以下内容: hadoop:2181 test kafka_streaming_topic 1

注意:

test:group名
1:线程数
setMaster("local[2]")   一定要大于2

7、run下代码,在kafka 生产者窗口手动输入几个单词,在kafka consumer窗口即时看到单词的产生,在本地代码的console窗口看到单词计数

  • Receiver方式的生产环境联调 1、在项目根目录下执行编译 mvn clean package -DskipTests 2、上传到服务器hadoop的lib目录下,执行:
spark-submit \
--class com.feiyue.bigdata.sparkstreaming.KafkaReceiverWordCount \
--master local[2] \
--name KafkaReceiverWordCount \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/hadoop/lib/spark-1.0-SNAPSHOT.jar hadoop:2181 test kafka_streaming_topic 1

3、运行后看4040端口Spark Streaming的UI界面

可以知道UI页面中,
Receiver是一直都在运作的,
而Direct方式没有此Jobs

Approach 2: Direct Approach (No Receivers)

Note that this feature was introduced in Spark 1.3 for the Scala and Java API, in Spark 1.4 for the Python API.

特点: 1、简化了并行度,不需要多个Input Stream,只需要一个DStream 2、加强了性能,真正做到了0数据丢失,而Receiver方式需要写到WAL才可以(即副本存储),Direct方式没有Receiver 3、只执行一次

缺点: 1、基于ZooKeeper的Kafka监控工具,无法展示出来,所以需要周期性地访问offset才能更新到ZooKeeper

  • 怎么做

基于Receiver方式的代码,将createStream改为createDirectStream,其余业务代码都不用改动。

    //kafkaParams: Map[String, String],
    //topics: Set[String]
    val Array(brokers, topics) = args


    //val sparkConf = new SparkConf().setAppName("KafkaDirectWordCount").setMaster("local[2]")
    val sparkConf = new SparkConf()

    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
    val topicsSet = topics.split(",").toSet

    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    messages.map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()

    ssc.start()
    ssc.awaitTermination()
  • Direct生产环境联调 基于Receiver方式,参数只需要传brokers与topics,注意查看源码与泛型看返回类型并构造出来
spark-submit \
--class com.feiyue.bigdata.sparkstreaming.KafkaDirectWordCount \
--master local[2] \
--name KafkaDirectWordCount \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/hadoop/lib/spark-1.0-SNAPSHOT.jar hadoop:9092  kafka_streaming_topic

3、运行后看4040端口Spark Streaming的UI界面

可以知道UI页面中,Direct方式没有此Jobs

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏C#

DotNet加密方式解析--非对称加密

    新年新气象,也希望新年可以挣大钱。不管今年年底会不会跟去年一样,满怀抱负却又壮志未酬。(不过没事,我已为各位卜上一卦,卦象显示各位都能挣钱...)...

5968
来自专栏大内老A

The .NET of Tomorrow

Ed Charbeneau(http://developer.telerik.com/featured/the-net-of-tomorrow/) Exciti...

38610
来自专栏张善友的专栏

Miguel de Icaza 细说 Mix 07大会上的Silverlight和DLR

Mono之父Miguel de Icaza 详细报道微软Mix 07大会上的Silverlight和DLR ,上面还谈到了Mono and Silverligh...

2997
来自专栏张善友的专栏

LINQ via C# 系列文章

LINQ via C# Recently I am giving a series of talk on LINQ. the name “LINQ via C...

3015
来自专栏一个会写诗的程序员的博客

Spring Reactor 项目核心库Reactor Core

Non-Blocking Reactive Streams Foundation for the JVM both implementing a Reactiv...

2802
来自专栏杨龙飞前端

scrollto 到指定位置

2954
来自专栏陈仁松博客

ASP.NET Core 'Microsoft.Win32.Registry' 错误修复

今天在发布Asp.net Core应用到Azure的时候出现错误InvalidOperationException: Cannot find compilati...

5238
来自专栏我和未来有约会

Kit 3D 更新

Kit3D is a 3D graphics engine written for Microsoft Silverlight. Kit3D was inita...

2936
来自专栏魂祭心

原 canvas绘制clock

5124
来自专栏张善友的专栏

Silverlight + Model-View-ViewModel (MVVM)

     早在2005年,John Gossman写了一篇关于Model-View-ViewModel模式的博文,这种模式被他所在的微软的项目组用来创建Expr...

3278

扫码关注云+社区