SparkStreaming和Kafka整合之路（最新版本）

2019独角兽企业重金招聘Python工程师标准>>>

最近完成了Spark Streaming和Kafka的整合工作&＃xff0c;耗时虽然不长&＃xff0c;但是当中还是遇到了不少的坑&＃xff0c;记录下来&＃xff0c;大家方便绕行。

先说一下环境&＃xff1a;

Spark 2.0.0 kafka_2.11-0.10.0.0

之前的项目当中&＃xff0c;已经在pom当中添加了需要的Spark Streaming的依赖&＃xff0c;这次只需要添加Spark Streaming Kafka的以来就行了&＃xff0c;问题来了。首先是我之前添加的Spark Streaming的依赖&＃xff1a;

      org.apache.spark
      spark-streaming_2.11
      2.0.0

然后是找到的spark streaming对kafka的支持依赖&＃xff1a;

    org.apache.spark
    spark-streaming-kafka_2.11
    1.6.2

请注意2个version部分&＃xff0c;好像差的有点多。不管了&＃xff0c;照着例子写写看&＃xff0c;果然报了各种class not found的错误。基本可以判断是版本差异造成的问题。

可是&＃xff0c;在http://mvnrepository.com上找不到更高版本的依赖怎么办呢&＃xff1f;

考虑了一下&＃xff0c;只有一个办法了&＃xff0c;下载spark源码&＃xff0c;自行编译打包需要的jar包。

在github上找到spark项目&＃xff0c;clone下来&＃xff0c;懒病又犯了&＃xff0c;也没仔细看当中的说明&＃xff0c;直接就clean compile等等。结果又是各种报错。好吧&＃xff0c;好好看看吧&＃xff0c;github上给了个地址&＃xff1a;http://spark.apache.org/docs/latest/building-spark.html&＃xff0c;照着做就没问题了。

然后把项目当中pom里面对streaming kafka的依赖删掉&＃xff0c;引入我们自己生成的jar包&＃xff1a;

spark-streaming-kafka-0-10_2.11-2.1.0-SNAPSHOT.jar

然后贴上代码&＃xff1a;

    val conf &＃61; new SparkConf().setAppName("kafkastream").setMaster("spark://master:7077").
      set("spark.driver.host", "192.168.1.142").
      setJars(List("/src/git/msgstream/out/artifacts/msgstream_jar/msgstream.jar",
        "/src/git/msgstream/lib/kafka-clients-0.10.0.0.jar",
        "/src/git/msgstream/lib/kafka_2.11-0.10.0.0.jar",
        "/src/git/msgstream/lib/spark-streaming-kafka-0-10_2.11-2.1.0-SNAPSHOT.jar"))
    val ssc &＃61; new StreamingContext(conf, Seconds(2))

    val topics &＃61; List("woozoom")
    val kafkaParams &＃61; Map(("bootstrap.servers", "master:9092,slave01:9092,slave02:9092"),
      ("group.id", "sparkstreaming"), ("key.deserializer", classOf[StringDeserializer]),
      ("value.deserializer", classOf[StringDeserializer]))
    val preferredHosts &＃61; LocationStrategies.PreferConsistent
    val offsets &＃61; Map(new TopicPartition("woozoom", 0) -> 2L)

    val lines &＃61; KafkaUtils.createDirectStream[String, String](
      ssc,
      preferredHosts,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets))

    lines.foreachRDD(rdd &＃61;> {
      rdd.foreach(x &＃61;> {
        println(x)
      })
    })

ssc.start()
ssc.awaitTermination()

上面标红的部分&＃xff0c;是需要注意的&＃xff0c;而这些本来我也是不会写的&＃xff0c;后来去到spark源码找到test代码

/src/git/spark/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala

测试&＃xff0c;通过&＃xff01;&＃xff01;&＃xff01;

总结&＃xff1a;

1、spark项目很多时候&＃xff0c;资源不是很充分&＃xff0c;想找例子的话&＃xff0c;2个途径&＃xff0c;一个spark安装包当中的example但是这个很多时候&＃xff0c;版本是比较老的&＃xff0c;不是很理想。更好地是从spark源码当中找他的测试用例&＃xff0c;这个基本上和你用的最新版本是完全匹配的。

2、编译过很多开源项目&＃xff0c;一般大的项目都会有相应的build说明&＃xff0c;照着那个做&＃xff0c;会为你节省很多时间。

3、从最开始遇到的版本号的问题来看&＃xff0c;很多时候我们遇到的问题并不一定是我们自己的问题&＃xff0c;不迷信&＃xff0c;大胆的相信自己的推测&＃xff0c;非常有助于问题的解决。