Spark kafka + streaming自适应topic partitions变化

背景

spark streaming + kafka 有两种方案接收kafka数据-基于receiver的方案和direct方案(no receiver方案)。

基于receiver的方案，属于比较老的方案，其采用Kafka’s high-level API通过专门的Rceiver去接收kafka数据。
采用 KafkaUtils.createStream
direct方案，是当前的主流用法，其采用Kafka’s simple consumer API，创建的RDD partitions数与kafka partitions数一致。性能比前者好。
采用 KafkaUtils.createDirectStream

具体的介绍可以查看官方介绍：Spark Streaming + Kafka Integration Guide

对于第二种方案，spark streaming启动后当kafka的topic partition数从A升到B时，新增的分区上的数据会丢失。
spark streaming只从topic中读取原来的A个分区数据，新增的分区并不能被spark streaming感知到。
更具体的原因是

DirectKafkaInputDStream#compute生成的KafkaRDD, 其partitions数与spark streaming启动时topic的partitions数一致，topic的partitions和offset保存在currentOffsets map变量中，
这个变量在启动时初始化，后续不会根据topic的partition变化进行更新。所以导致kafka新增的partitions数据
会丢失。

更详细的分析，可以参考下面的文章：
Spark Streaming 自适应上游 kafka topic partition 数目变化

因为spark streaming是24x7运行的，如何让spark streaming不重启的情况下，自适应topic partitions的变化？
且能够符合如下要求:

不需要重新编译spark源码, 因为源代码不是我们维护的。
能够灵活的部署

解决方案

思路可以参考
Spark Streaming 自适应上游 kafka topic partition 数目变化

总的解决方案如下：

MTDirectKafkaInputDStream继承DirectKafkaInputDStream，override compute方法，在每次生成KafkaRDD时，更新currentOffsets中的分区信息。
在org.apache.spark.streaming.kafka路径下,新建一个KafkaUtils.scala文件，里面的代码直接将spark源码中的KafkaUtils源码复制过来。修改新建的KafkaUtils.scala，将createDirectStream中new DirectKafkaInputDStream，替换为 new MTDirectKafkaInputDStream.

具体的实现步骤如下

新建maven工程 kafkastreamingadpter, 配置好scala相关的pom配置。
新建package org.apache.spark.streaming.kafka

备注：这里需要说明，因为下面创建的MTDirectKafkaInputDStream需要继承DirectKafkaInputDStream，而DirectKafkaInputDStream是private[streaming]修饰的。这里的规避技巧就是，将需要继承DirectKafkaInputDStream的子类对应的package设置为和DirectKafkaInputDStream一致，即可规避private无法继承的问题。
新建MTDirectKafkaInputDStream继承DirectKafkaInputDStream

override compute方法，在每次生成KafkaRDD时，更新currentOffsets中的分区信息。

在org.apache.spark.streaming.kafka路径下,新建一个KafkaUtils.scala文件，里面的代码直接复制spark源码中的KafkaUtils源码。修改新建的KafkaUtils.scala，将createDirectStream中new DirectKafkaInputDStream，替换为 new MTDirectKafkaInputDStream

备注：streaming代码中是通过 KafkaUtils.createDirectStream来创建stream的。因此希望KafkaUtils.createDirectStream返回的是MTDirectKafkaInputDStream。所以要修改KafkaUtils代码，将createDirectStream中new DirectKafkaInputDStream，修改为 new MTDirectKafkaInputDStream。
我们不希望修改spark源码，所以这里用来个小技巧，直接将KafkaUtils代码copy一份出来进行修改，需要保持KafkaUtils的package路径一致

2,4中用到的小技巧参考了

[2016中国云计算技术大会-腾讯林立伟-Spark-Streaming在腾讯广点通的应用.pdf]

Spark Streaming 自适应上游 kafka topic partition 数目变化
新建spark streaming + kafka应用实例DirectKafkaWordCount1
实例需要配置 spark、kafka streaming相关的依赖。

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>

mvn clean package 打包生成kafka-streaming-adpter-1.0.jar
对应的配置文件如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.sensetime.iva</groupId>
    <artifactId>kafka-streaming-adpter</artifactId>
    <version>1.0</version>
    <packaging>jar</packaging>
    <properties>
        <scala.version>2.11.8</scala.version>
        <commons.codec.version>1.8</commons.codec.version>
        <grizzled.version>1.0.1</grizzled.version>
        <slf4j-log4j12.version>1.7.5</slf4j-log4j12.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>${commons.codec.version}</version>
        </dependency>
        <dependency>
            <groupId>org.clapper</groupId>
            <artifactId>grizzled-slf4j_2.10</artifactId>
            <version>${grizzled.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j-log4j12.version}</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

上传jar包到spark on k8s的根路径下的jars，进行镜像打包

node 73节点
cd /home/huangchuibi/k8s/spark-2.2.0-k8s-0.5.0-bin-2.7.3
./sbin/build-push-docker-images.sh -r register.sensetime.sz.test-portus.com/spark-on-k8s -t 0.1 build
./sbin/build-push-docker-images.sh -r register.sensetime.sz.test-portus.com/spark-on-k8s -t 0.1 push

在k8s各个节点上pull 镜像，因为spark on k8s，配置镜像拉取策略为本地有就不再拉取。所以需要手动进行拉取。

docker pull register.sensetime.sz.test-portus.com/spark-on-k8s/spark-driver:0.1
docker pull register.sensetime.sz.test-portus.com/spark-on-k8s/spark-executor:0.1

创建topic

ssh node91
kubectl -n snappydata exec -it kafka-0 /bin/bash
cd /opt/kafka_2.11-0.10.2.0
bin/kafka-topics.sh --create --zookeeper zookeeper-0.zookeeper.snappydata.svc.cluster.local:2181,zookeeper-1.zookeeper.snappydata.svc.cluster.local:2181,zookeeper-2.zookeeper.snappydata.svc.cluster.local:2181/kafka --replication-factor 1 --partitions 1 --topic test8

运行streaming 应用

bin/spark-submit \
  --deploy-mode cluster \
  --class org.apache.spark.examples.streaming.DirectKafkaWordCount1 \
  --master k8s://http://172.20.2.91:8001 \
  --kubernetes-namespace snappydata \
  --conf spark.executor.instances=3 \
  --conf spark.app.name=spark-streaming \
  --conf spark.kubernetes.driver.docker.image=register.sensetime.sz.test-portus.com/spark-on-k8s/spark-driver:0.1 \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  --conf spark.kubernetes.executor.docker.image=register.sensetime.sz.test-portus.com/spark-on-k8s/spark-executor:0.1 \
  --conf spark.kubernetes.initcontainer.docker.image=register.sensetime.sz.test-portus.com/spark-on-k8s/spark-init:0.1 \
  --conf "spark.driver.extraJavaOptions=-Xss10m" \
  --conf "spark.executor.extraJavaOptions=-Xss10m" \
  --conf "spark.driver.extraClassPath=/opt/spark/examples/jars/kafka-streaming-adpter-1.0.jar" \
  --conf "spark.executor.extraClassPath=/opt/spark/examples/jars/kafka-streaming-adpter-1.0.jar" \
  --conf "spark.executor.userClassPathFirst=true" \
  --conf "spark.driver.userClassPathFirst=true" \
  local:///opt/spark/examples/jars/kafka-streaming-adpter-1.0.jar kafka-0.kafka.snappydata.svc.cluster.local:9092,kafka-1.kafka.snappydata.svc.cluster.local:9092,kafka-2.kafka.snappydata.svc.cluster.local:9092 test8

备注这里增加的四个conf配置，是为了让kafka-streaming-adpter-1.0.jar里的KafkaUtils优先级高于spark源码中的KafkaUtils。但是实际测试发现，应用还是以同一个jar包中的KafkaUtils为优先级最高。所以下面的四个配置也可以去掉。

--conf "spark.driver.extraClassPath=/opt/spark/examples/jars/kafka-streaming-adpter-1.0.jar" \
--conf "spark.executor.extraClassPath=/opt/spark/examples/jars/kafka-streaming-adpter-1.0.jar" \
--conf "spark.executor.userClassPathFirst=true" \
--conf "spark.driver.userClassPathFirst=true" \

动态增加分区

kafka-topics.sh --zookeeper zookeeper-0.zookeeper.snappydata.svc.cluster.local:2181,zookeeper-1.zookeeper.snappydata.svc.cluster.local:2181,zookeeper-2.zookeeper.snappydata.svc.cluster.local:2181/kafka -alter -partitions 2 --topic test8