Spark Streaming对接Kafka0-10样例程序（Scala）

功能介绍

在Spark应用中，通过使用Streaming调用Kafka接口来获取单词记录，然后把单词记录分类统计，得到每个单词记录数，或将数据写入Kafka0-10。

Streaming读取Kafka0-10代码样例

下面代码片段仅为演示，具体代码参见：com.huawei.bigdata.spark.examples.SecurityKafkaWordCount。

/**
  *从Kafka的一个或多个主题消息。
  * <checkPointDir>是Spark Streaming检查点目录。
  * <brokers>是用于自举，制作人只会使用它来获取元数据
  * <topics>是要消费的一个或多个kafka主题的列表
  * <batchTime>是Spark Streaming批次持续时间（以秒为单位）。
  */
object SecurityKafkaWordCount {

  def main(args: Array[String]) {
    val ssc = createContext(args)

    //启动Streaming系统。
    ssc.start()
    ssc.awaitTermination()
  }

  def createContext(args : Array[String]) : StreamingContext = {






    val Array(checkPointDir, brokers, topics, batchSize) = args

    //新建一个Streaming启动环境。
    val sparkConf = new SparkConf().setAppName("KafkaWordCount")
    val ssc = new StreamingContext(sparkConf, Seconds(batchSize.toLong))

    //配置Streaming的CheckPoint目录。
    //由于窗口概念的存在，此参数是必需的。
    ssc.checkpoint(checkPointDir)

    //获取kafka使用的topic列表。
    val topicArr = topics.split(",")
    val topicSet = topicArr.toSet
    val kafkaParams = Map[String, String](
      "bootstrap.servers" -> brokers,
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "DemoConsumer",
      "security.protocol" -> "SASL_PLAINTEXT",
      "sasl.kerberos.service.name" -> "kafka",
      "kerberos.domain.name" -> "hadoop.<系统域名>"
    );

    val locationStrategy = LocationStrategies.PreferConsistent
    val consumerStrategy = ConsumerStrategies.Subscribe[String, String](topicSet, kafkaParams)

    // 用brokers and topics新建direct kafka stream
    //从Kafka接收数据并生成相应的DStream。
    val stream = KafkaUtils.createDirectStream[String, String](ssc, locationStrategy, consumerStrategy)

    //获取每行中的字段属性。
    val tf = stream.transform ( rdd =>
      rdd.map(r => (r.value, 1L))
    )

    //汇总计算字数的总时间。
    val wordCounts = tf.reduceByKey(_ + _)
    val totalCounts = wordCounts.updateStateByKey(updataFunc)
    totalCounts.print()
    ssc
  }

  def updataFunc(values : Seq[Long], state : Option[Long]) : Option[Long] =
    Some(values.sum + state.getOrElse(0L))
}

Streaming Write To Kafka 0-10样例代码

下面代码片段仅为演示，具体代码参见com.huawei.bigdata.spark.examples.DstreamKafkaWriter。

建议使用新的API createDirectStream代替原有API createStream进行应用程序开发。原有API仍然可以使用，但新的API性能和稳定性更好。