在阅读多分区 kafka 主题时，如何启动和调用 spark 结构化流式消费者？

Question

如果一个 kakfa 主题有多个分区，在 java 中，那些许多消费者实例/线程将在消费者端实例化。

如何在火花流消费者方面处理它？我没有找到很多关于相同的信息。相同的任何示例，即在主题的 spark-streaming-consumer 调用多个消费者。

非常感谢任何设计建议/示例。

问候， 夏姆

Answer 1

如果 Kafka 有多个分区，这意味着消费者可以通过并行执行某项任务从中受益。特别是 spark-streaming 在内部可以通过增加 num-executors 参数来加速作业。这与 Kafka 拥有的分区数量有关，例如如果您的 Kafka 分区数量与 spark 中的 num-executors 数量相同，理论上所有 executor 都可以一次读取所有分区，这显然提高系统吞吐量。

Answer 2

只要 spark 有足够的资源，Spark 流总是从 Kafka 中的所有可用分区并行读取数据。 Spark 开箱即用，我们不需要为此编写任何代码。

例如，如果您的 Kafka 主题有 4 个分区，那么如果您启动 你的火花作业有 2 个执行器，每个执行器有 2 个核心，然后是你的火花作业 将启动 4 个任务从 4 个 Kafka 并行读取数据 分区。

如果您需要更多信息，请随时发表评论。

https://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html

import java.sql.Timestamp

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import java.time.{LocalDate, LocalDateTime}
import java.util.Calendar



object SparkKafka {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("test_app")
      .getOrCreate()
    val sparkContext = spark.sparkContext


    val ssc = new StreamingContext(sparkContext, Seconds(1)) // the polling frequency is 2 seconds, can be modified based on the BM requirements.
///val currentHour = now.get(Calendar.HOUR_OF_DAY)

    log.info("Before starting the Stream -->>")
    val stream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String]
      (Array.apply("Kafka_topic_name"), getKafkaParams()))

      .map(record => record.value)

    stream.foreachRDD { rdd =>

      try {
        if (!rdd.isEmpty()) {
          log.info("rdd is not empty and saving to -->>"+LocalDate.now.getYear+"/"+LocalDate.now.getMonth+"/"+LocalDate.now.getDayOfMonth+"/"+LocalDateTime.now().getHour)
          rdd.saveAsTextFile("hdfs:///<folder to save>") //TODO::: Externalize the HDFS location to Props




          LocalDate.now.getMonth


         if (null != args && null != args {
            0
          } && args {
            0
          }.equals("log")) {
            rdd.foreach(x => print("Message read and saved TO S3 bucket----*****--->>" + x))
          }
        }
      } catch {

        case t: Throwable =>
          t.printStackTrace() // TODO: handle error)
          log.error("Exception occured while processing the data exception is {}", t.getCause)
      }
    }

    ssc.start()
    log.info("started now-->> " + compat.Platform.currentTime)
    ssc.awaitTermination()

  }

  def getKafkaParams(): Map[String, Object] = {
    Map[String, Object]("bootstrap.servers" -> "host:port
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "Group_Name",
      //      "sasl.kerberos.service.name" -> "kafka",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean))
  }

}