spark streaming 交互

Apache Spark 是一个高性能集群计算框架，其中 Spark Streaming 作为实时批处理组件，因为其简单易上手的特性深受喜爱。在 es-hadoop 2.1.0 版本之后，也新增了对 Spark 的支持，使得结合 ES 和 Spark 成为可能。

目前最新版本的 es-hadoop 是 2.1.0-Beta4。安装如下：

  1. wget http://d3kbcqa49mib13.cloudfront.net/spark-1.0.2-bin-cdh4.tgz
  2. wget http://download.elasticsearch.org/hadoop/elasticsearch-hadoop-2.1.0.Beta4.zip

然后通过 ADD_JARS=../elasticsearch-hadoop-2.1.0.Beta4/dist/elasticsearch-spark_2.10-2.1.0.Beta4.jar 环境变量，把对应的 jar 包加入 Spark 的 jar 环境中。

下面是一段使用 spark streaming 接收 kafka 消息队列，然后写入 ES 的配置：

  1. import org.apache.spark._
  2. import org.apache.spark.streaming.kafka.KafkaUtils
  3. import org.apache.spark.streaming._
  4. import org.apache.spark.streaming.StreamingContext._
  5. import org.apache.spark.SparkContext
  6. import org.apache.spark.SparkContext._
  7. import org.apache.spark.SparkConf
  8. import org.apache.spark.sql._
  9. import org.elasticsearch.spark.sql._
  10. import org.apache.spark.storage.StorageLevel
  11. import org.apache.spark.Logging
  12. import org.apache.log4j.{Level, Logger}
  14. object Elastic {
  15.   def main(args: Array[String]) {
  16.     val numThreads = 1
  17.     val zookeeperQuorum = "localhost:2181"
  18.     val groupId = "test"
  19.     val topic = Array("test").map((_, numThreads)).toMap
  20.     val elasticResource = "apps/blog"
  22.     val sc = new SparkConf()
  23.                  .setMaster("local[*]")
  24.                  .setAppName("Elastic Search Indexer App")
  26.     sc.set("es.index.auto.create", "true")
  27.     val ssc = new StreamingContext(sc, Seconds(10))
  28.     ssc.checkpoint("checkpoint")
  29.     val logs = KafkaUtils.createStream(ssc,
  30.                                        zookeeperQuorum,
  31.                                        groupId,
  32.                                        topic,
  33.                                        StorageLevel.MEMORY_AND_DISK_SER)
  34.                          .map(_._2)
  36.     logs.foreachRDD { rdd =>
  37.       val sc = rdd.context
  38.       val sqlContext = new SQLContext(sc)
  39.       val log = sqlContext.jsonRDD(rdd)
  40.       log.saveToEs(elasticResource)
  41.     }
  43.     ssc.start()
  44.     ssc.awaitTermination()
  46.   }
  47. }

注意，代码中使用了 spark SQL 提供的 jsonRDD() 方法，如果在对应的 kafka topic 里的数据，本身并不是已经处理好了的 JSON 数据的话，这里还需要自己写一写额外的处理函数，利用 cast class 来规范数据。