Spark Streaming读写Kafka，将offset写入

作者: xiaogp | 来源:发表于2020-09-29 14:40 被阅读0次

Spark Streaming读写Kafka，将offset写入
【offset管理策略】Spark Streaming消费Kaf
如何管理Spark Streaming消费Kafka的偏移量（一
Spark Streaming和Kafka集成深入浅出
SparkStreaming写Hbase速度提升5倍
sparkStreaming读取kafka中数据时，如何有效的对
Kafka学习
ES-HADOOP配置
kafka.consumer.SimpleConsumer使用案
spark streaming 集成kafka0.10 offs

摘要： Spark Streaming， Kafka，zookeeper

Scala调用Kafka API发送数据

使用Scala调用Kafka API向Kafka发送数据，设置topic，key/value的序列化方式，创建一个KafkaProducer对象和多个ProducerRecord不断调用send方法。

import java.util.Properties

import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.serialization.StringSerializer

import scala.util.Random

object Producer {
  val topic = "test_gp"
  val brokers = "cloudera01:9092"
  val rnd = new Random()
  val prop = new Properties() {
    {
      put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
      put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
      put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    }
  }

  def main(args: Array[String]): Unit = {
    val producer = new KafkaProducer[String, String](prop)
    // 模拟数据
    val nameAddres = Map("bob" -> "shanghai#200000", "amy" -> "beijing#100000", "alice" -> "shanghai#200000", "tom" -> "beijing#100000", "lulu" -> "hangzhou#310000",
      "nick" -> "shanghai#200000")
    val namePhones = Map("bob" -> "15700079421", "amy" -> "18700079458", "alice" -> "17730079427", "tom" -> "16700379451", "lulu" -> "18800074423",
      "nick" -> "14400033426")

    for (nameAddre <- nameAddres) {
      val data = new ProducerRecord[String, String](topic, nameAddre._1, s"${nameAddre._1}\t${nameAddre._2}\t0")
      producer.send(data, new Callback {
        override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {
          if (null != e) e.printStackTrace()
        }
      })
      if (rnd.nextInt(100) < 50) Thread.sleep(rnd.nextInt(10))
    }

    for (namePhone <- namePhones) {
      val data = new ProducerRecord[String, String](topic, namePhone._1, s"${namePhone._1}\t${namePhone._2}\t0")
      producer.send(data, new Callback {
        override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {
          if (null != e) e.printStackTrace()
        }
      })
      if (rnd.nextInt(100) < 50) Thread.sleep(rnd.nextInt(10))
    }

    producer.close()
  }
}

使用kafka客户端查看接受到的数据
test_gp有5个主题，2个备份，分布在3台机器上

[root@cloudera01 ~]# kafka-topics --describe --zookeeper cloudera01:2181/kafka --topic test_gp
Topic:test_gp   PartitionCount:5    ReplicationFactor:2 Configs:
    Topic: test_gp  Partition: 0    Leader: 79  Replicas: 79,78 Isr: 78,79
    Topic: test_gp  Partition: 1    Leader: 77  Replicas: 77,79 Isr: 77,79
    Topic: test_gp  Partition: 2    Leader: 78  Replicas: 78,77 Isr: 78,77
    Topic: test_gp  Partition: 3    Leader: 79  Replicas: 79,77 Isr: 77,79
    Topic: test_gp  Partition: 4    Leader: 77  Replicas: 77,78 Isr: 78,77
[root@cloudera01 ~]# kafka-console-consumer --bootstrap-server cloudera01:9092 --topic test_gp
nick    shanghai#200000 0
lulu    hangzhou#310000 0
bob shanghai#200000 0
tom beijing#100000  0
amy beijing#100000  0
nick    14400033426 1
lulu    18800074423 1
bob 15700079421 1
tom 16700379451 1
amy 18700079458 1
alice   shanghai#200000 0
alice   17730079427 1

Spark Streaming消费Kafka数据并写入Kafka

从kafka中抽取数据进行消费，将原始数据解析成json格式写入一个新的kafka，test_gp2，通过客户端查看test_gp2有两个分区

[root@cloudera01 ~]# kafka-topics --zookeeper cloudera01:2181/kafka --describe --topic test_gp2
Topic:test_gp2  PartitionCount:2    ReplicationFactor:3 Configs:
    Topic: test_gp2 Partition: 0    Leader: 78  Replicas: 78,79,77  Isr: 78,77,79
    Topic: test_gp2 Partition: 1    Leader: 79  Replicas: 79,77,78  Isr: 78,77,79

开发步骤：

SparkConf创建spark配置
StreamingContext创建spark streaming上下文
创建Kafka配置参数kafkaParams: Map[String, Object]
创建其他配置并且使用ssc上下文广播
通过KafkaUtils.createDirectStream获取kafkaDirectStream DStream对象
map转换操作将String转化为JSONObject
foreachRDD输出，rdd调用foreach输出每条消息到kafka，在fearch内部调用KafkaProducer单例对象，如果KafkaProducer异步发送消息成功将offset写入zookeeper
先创建一个KafkaUtilities脚本构建KafkaProducer单例对象，否则每个partition中每条记录都需要新建一个KafkaProducer增加了不必要的开销。

import java.util.Properties
import org.apache.kafka.clients.producer.{ Producer, ProducerRecord, KafkaProducer }
import org.apache.kafka.clients.producer.Callback

object KafkaUtilities {
  private var producer: Producer[String, String] = null

  def getProducer(bootstrap: String): Producer[String, String] = {
    this.synchronized {
      if (producer == null) {
        producer = createKafkaProducer(bootstrap)
      }
      producer
    }
  }

  def createKafkaProducer(bootstrap: String): Producer[String, String] = {
    val props = new Properties()
    props.put("max.request.size", "50000000")
    props.put("buffer.memory", "50000000")
    props.put("timeout.ms", "3000000")
    props.put("request.timeout.ms", "30000000")
    props.put("socket.request.max.bytes", "60000000")
    props.put("message.max.bytes", "60000000")
    props.put("replica.fetch.max.bytes", "100000000")
    props.put("fetch.message.max.bytes", "100000000")
    props.put("serializer.class", "kafka.serializer.StringEncoder")
    props.put("request.required.acks", "1")
    props.put("max.in.flight.requests.per.connection", "1")
    props.put("retries", "3")
    props.put("bootstrap.servers", bootstrap)
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("max.request.size", "50000000")
    props.put("buffer.memory", "50000000")
    props.put("timeout.ms", "3000000")
    props.put("request.timeout.ms", "30000000")
    props.put("socket.request.max.bytes", "60000000")
    props.put("message.max.bytes", "60000000")
    props.put("replica.fetch.max.bytes", "100000000")
    props.put("fetch.message.max.bytes", "100000000")

    new KafkaProducer[String, String](props)
  }

  /**
   * 异步发送消息到kafka
   * @param kafkaProducer
   * @param topic
   * @param message
   * @param key
   * @param partition
   * @param callback
   * @return
   */
  def sendKafka(
         kafkaProducer: Producer[String, String],
         topic:         String,
         message:       String,
         key:           String,
         partition:     Integer,
         callback:      Callback) = {
    if (callback != null && callback.isInstanceOf[Callback]) {
      kafkaProducer.send(new ProducerRecord[String, String](topic, partition, key, message), callback)
    } else {
      kafkaProducer.send(new ProducerRecord[String, String](topic, partition, key, message))
    }
  }
}

在send方法中手动指定了key和partition，如果一个有效的partition属性数值被指定，那么在发送记录时partition属性数值就会被应用发到指定的分区编号。如果没有partition属性数值被指定，而一个key属性被声明的话，一个partition会通过key的hash而被选中发送到对应的partition。如果既没有key也没有partition属性数值被声明，那么一个partition将会被分配以轮询的方式。

创建工厂模式zookeeper方法，实现往zookeeper写入数据


import org.apache.zookeeper._
import org.apache.zookeeper.common._
import java.util.Properties

import org.apache.zookeeper.data.Stat

class ZKUtils() extends Watcher {
  private var zk: ZooKeeper = _
  private var pro: Properties = _

  private def this(pro: Properties) = {
    this()
    this.pro = pro
    this.zk = new ZooKeeper(pro.getProperty("ZK_HOST"), 20000, this)
  }

  @Override
  def process(event: WatchedEvent) {
    //no watching
  }

  private def createPath(path: String, zk: ZooKeeper): Unit = {
    //创建路径
    PathUtils.validatePath(path)
    var parentPath = path.substring(0, path.indexOf("/", 0) + 1)
    while (parentPath != null && !parentPath.isEmpty) {
      if (zk.exists(parentPath, false) == null) {
        // 同步创建，acl权限完全开放，节点类型持久化节点
        zk.create(parentPath, "".getBytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      }

      // 如果根路径和全路径一致停止创建
      if (parentPath.equals(path)) {
        parentPath = null
      } else {
        // 寻找下一个/作为新根路径
        var idx = path.indexOf("/", parentPath.length() + 1)
        if (-1 == idx) {
          idx = path.length
        }
        parentPath = path.substring(0, idx)
      }
    }
  }

  def setData(path: String, data: String): Stat = {
    if (zk.exists(path, false) == null) {
      createPath(path, zk)
    }
    zk.setData(path, data.getBytes, -1)
  }

  def getData(path: String): String = {
    val stat = zk.exists(path, false)
    if (stat == null) {
      null
    } else {
      new String(zk.getData(path, false, stat))
    }
  }
}


object ZKUtils {
  private var ZKUtils: ZKUtils = _

  def getIns(prop: Properties): ZKUtils = {
    this.synchronized {
      if (ZKUtils == null) {
        ZKUtils = new ZKUtils(prop)
      }
      ZKUtils
    }
  }
}

消费test_gp消息，每条消息调用KafkaProducer单例对象写入test_gp2

import java.util.Properties

import org.apache.log4j.{Level, Logger}
import com.alibaba.fastjson.JSONObject
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{Callback, RecordMetadata}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.{Seconds, StreamingContext}


object KafkaOperation {
  Logger.getRootLogger.setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {
    // spark配置
    val sparkConf = new SparkConf()
      .setAppName("KafkaOperation")
      .setMaster("local[*]")
      .set("spark.streaming.kafka.maxRatePerPartition", "100") // 每秒对一个topic下每个partition拉取的最大数据量
    // 这个参数需要和batchDuration一起调节做到拉取和处理数据平衡
    // 默认没有上限，在一个batchDuration下kafka有多少数据就拉取多少数据，由于batchDuration时间不能动态调节，所以当数据流入量大了可能会堵塞

    // 创建上下文
    val ssc = new StreamingContext(sparkConf, Seconds(2))

    // 其他配置广播
    val properties: Properties = new Properties()
    properties.setProperty("bootstrap.servers", "cloudera01:9092")
    properties.setProperty("kafka.topic.save", "test_gp2")
    properties.setProperty("kafka.topic.input", "test_gp")
    properties.setProperty("group.id", "kafka_operation_group")
    properties.setProperty("ZK_HOST", "cloudera01,cloudera02,cloudera03")
    val configProperties = ssc.sparkContext.broadcast(properties)

    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "cloudera01:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> "kafka_operation_group",
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )

    // 获取kafka DStream
    val kafkaDirectStream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent, // 在executor上均匀分布分区
      Subscribe[String, String](List("test_gp"), kafkaParams)) // 订阅主题集合，kafka参数

    val saveRDD = kafkaDirectStream.map(x => {
      // 添加partition和offset到数据
      val jsonObject = new JSONObject() {
        {
          put("partition", x.partition().toString)
          put("offset", x.partition().toString)
        }
      }
      val Array(name, tmp, label) = x.value().split("\t")
      jsonObject.put("name", name)
      if (label.toInt == 0) jsonObject.put("addr", tmp) else jsonObject.put("phone", tmp)
      jsonObject
    })

    // 写入kafka
    saveRDD.foreachRDD(r => {
      if (r != null && !r.isEmpty()) {
        r.foreach(x => {
          val name = x.getString("name")
          try {
            KafkaUtilities.sendKafka(
              KafkaUtilities.getProducer(configProperties.value.getProperty("bootstrap.servers")), // KafkaProducer单例
              configProperties.value.getProperty("kafka.topic.save"),
              x.toJSONString,
              name,
              new scala.util.Random().nextInt(2),
              new Callback {
                override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {
                  if (recordMetadata != null && e == null) {
                    //保存偏移量到 zookeeper
                    ZKUtils.getIns(configProperties.value).setData(
                      "/test/monitor/" + configProperties.value.getProperty("kafka.topic.input") + "/" +
                        configProperties.value.getProperty("group.id") + "/" +
                        x.getString("partition"),
                      x.getString("offset"))
                  } else {
                    println("异步发送消息失败", x.toJSONString)
                    e.printStackTrace()
                  }
                }
              }
            )
          }
        })
      }
    })

    // 开始运算
    ssc.start()
    // 等待被打断停止
    ssc.awaitTermination()
  }
}

查看kafka收到的消息

[root@cloudera01 ~]# kafka-console-consumer --bootstrap-server cloudera01:9092 --topic test_gp2
{"partition":"2","offset":"108","phone":"14400033426","name":"nick"}
{"partition":"2","offset":"106","name":"nick","addr":"shanghai#200000"}
{"partition":"2","offset":"107","name":"lulu","addr":"hangzhou#310000"}
{"partition":"2","offset":"109","phone":"18800074423","name":"lulu"}
{"partition":"1","offset":"107","name":"amy","addr":"beijing#100000"}
{"partition":"3","offset":"53","name":"bob","addr":"shanghai#200000"}
{"partition":"3","offset":"54","phone":"15700079421","name":"bob"}
{"partition":"1","offset":"106","name":"tom","addr":"beijing#100000"}
{"partition":"1","offset":"108","phone":"16700379451","name":"tom"}
{"partition":"1","offset":"109","phone":"18700079458","name":"amy"}
{"partition":"4","offset":"53","name":"alice","addr":"shanghai#200000"}
{"partition":"4","offset":"54","phone":"17730079427","name":"alice"}

查看zookeeper存储的offset信息

[zk: localhost:2181(CONNECTED) 30] ls /test/monitor/test_gp/kafka_operation_group
[1, 2, 3, 4]
[zk: localhost:2181(CONNECTED) 18] get /test/monitor/test_gp/kafka_operation_group/1
109
cZxid = 0x3d000877d5
ctime = Wed Sep 30 14:40:46 CST 2020
mZxid = 0x3d00087fb1
mtime = Wed Sep 30 15:32:18 CST 2020
pZxid = 0x3d000877d5
cversion = 0
dataVersion = 38
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

Spark Streaming读写Kafka，将offset写入
摘要： Spark Streaming， Kafka，zookeeper Scala调用Kafka API发送数据...
【offset管理策略】Spark Streaming消费Kaf
摘要：offset管理，Spark Streaming，Kafka Spark Streaming offset的...
如何管理Spark Streaming消费Kafka的偏移量（一
本篇我们先从理论的角度聊聊在Spark Streaming集成Kafka时的offset状态如何管理。 spark...
Spark Streaming和Kafka集成深入浅出
写在前面本文主要介绍Spark Streaming基本概念、kafka集成、Offset管理本文主要介绍Spa...
SparkStreaming写Hbase速度提升5倍
业务背景：使用Spark 、streaming从kafka读取数据后写入HBase。kafkaDStream是从k...
sparkStreaming读取kafka中数据时，如何有效的对
在spark streaming读取kafka的数据中，spark streaming提供了两个接口读取kafka...
Kafka学习
1. Kafka消费方式：通过Flume，将kafka消息存储到HDFS 通过Spark Streaming读取...
ES-HADOOP配置
参考使用 ES-Hadoop 将 Spark Streaming 流数据写入 ES
kafka.consumer.SimpleConsumer使用案
摘要：SimpleConsumer，Zookeeper，offset管理，Spark Streaming Simp...
spark streaming 集成kafka0.10 offs
spark streaming+kafka 0.10集成默认spark partition和kafka pa...