SparkStreaming与kafka的结合

2020-12-05 18:04:12 阅读：198 来源： 互联网

标签：String 9092 SparkStreaming 结合 value key apache kafka

首先,启动zookeeper,kafka,
启动生产者

kafka-console-producer.sh \
--broker-list mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

启动消费者

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

编写代码,并执行

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamKafaka extends App {
  private val conf = new SparkConf().setAppName("test").setMaster("local[*]")
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  private val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))
  private val paras: Map[String, String] = Map[String, String](
    "bootstrap.servers" -> "mypc01:9092,mypc02:9092,mypc03:9092",
    "group.id" -> "test1",
    "auto.offset.restet" -> "latest",
    "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
    "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
  )
  //利用kafka构建dstream
  //3个参数,sparkconf,xx,消费者策略
  private val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
    ConsumerStrategies.Subscribe(Array("pet"), paras))
  dstream.print()
  ssc.start()
  ssc.awaitTermination()
}

之后在生产者生产消息,idea的输出控制台就可以看到输出了如下示例

ConsumerRecord(topic = pet, partition = 2, offset = 185, CreateTime = 1607160465281, checksum = 314762068, serialized key size = -1, serialized value size = 3, key = null, value = 999)
ConsumerRecord(topic = pet, partition = 0, offset = 159, CreateTime = 1607160469953, checksum = 4250414327, serialized key size = -1, serialized value size = 5, key = null, value = apple)
ConsumerRecord(topic = pet, partition = 1, offset = 175, CreateTime = 1607160467851, checksum = 2772696894, serialized key size = -1, serialized value size = 4, key = null, value = ghjk)

其中获得的对象还可以进一步解析,比如你只想打印value

dstream.map((x: ConsumerRecord[String, String]) =>{
    x.value()
  }).print()

方法解析

createDirectStream方法

def createDirectStream[K, V](
      ssc: StreamingContext,
      locationStrategy: LocationStrategy,
      consumerStrategy: ConsumerStrategy[K, V]
    )

DStream的 Scala构造函数，其中每个给定的Kafka主题/分区都对应于RDD分区。 spark配置spark.streaming.kafka.maxRatePerPartition给出每个分区每秒接受的最大消息数。
参数：
locationStrategy –在大多数情况下，请传入LocationStrategies.PreferConsistent，有关更多详细信息，请参见LocationStrategies。
ConsumerStrategy –在大多数情况下，请传入ConsumerStrategies。请订阅，请参阅ConsumerStrategies以获取更多详细信息
类型参数：
K – Kafka消息key的类型
V – Kafka消息value的类型

Subscribe方法

def Subscribe[K, V](topics: Iterable[String], 
					kafkaParams: Map[String, Object]): ConsumerStrategy[K, V]

:订阅主题集合。
参数：
主题–订阅主题的集合
kafkaParams –在驱动程序上使用的Kafka配置参数。执行者将使用相同的参数，并进行少量的自动修改。需要使用以host1：port1，host2：port2格式指定的Kafka代理设置“ bootstrap.servers”。

标签：String,9092,SparkStreaming,结合,value,key,apache,kafka
来源： https://blog.csdn.net/u010711495/article/details/110704235

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

SparkStreaming与kafka的结合

方法解析

createDirectStream方法

Subscribe方法