码上快乐

相关内容   简体   繁体

kafka+SparkStreaming以及Kafka+Kerberos+Sentry+SparkStreaming代码调试问题与分析

本文转载自  查看原文  2017-12-07 10:18  1944    spark/ kafka/ BigData/ sparkstreaming/ kerberos/ sentry
【调试背景】
目前测试kafka集群有两套，版本为 0.10.x。有一套是添加了Kerberos+Sentry认证，另一套没有添加。
现在需要通过sparkStreaming接入kafka做实时分析。
【总体结论】
实验1：1.6.x版本spark的jar包，0.8.x.x的spark-streaming-kafka，无Kerberos+Sentry认证，用createStream，可以从zk中获取broker，接入成功；
实验2：1.6.x版本spark的jar包，0.8.x.x的spark-streaming-kafka，有Kerberos+Sentry认证，用createStream，无法zk中获取broker，接入失败，报空指针；
实验3：1.6.x版本spark的jar包，0.8.x.x的spark-streaming-kafka，有Kerberos+Sentry认证，用createDirectStream，直接设置broker，接入失败，报EOFException；
实验4：2.1.x版本spark的jar包，010版本的spark-streaming-kafka，有Kerberos+Sentry认证，用createDirectStream，直接设置broker，需要修改“KafkaUtils”源码，接入成功；
PS     ：2.1.x版本spark的jar包，010版本的spark-streaming-kafka，无Kerberos+Sentry认证，用createDirectStream，直接设置broker，接入成功；
【实验1】可以正常运行
（1）kafka环境：无Kerberos+Sentry认证
（2）使用jar包：

  
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>1.6.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.10</artifactId>
            <version>1.6.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.10</artifactId>
            <version>1.6.1</version>
        </dependency>
（3）核心代码： 
package com.xx.kafka; import java.util.HashMap; import java.util.Map; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.Function; import org.apache.spark.streaming.Duration; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import org.apache.spark.streaming.kafka.KafkaUtils; import scala.Tuple2; public class PrintRecMsg { public static void main(String[] args) { Map<String, Integer> topicmap = new HashMap<>(); topicmap.put(args[0], 2); SparkConf sparkConf = new SparkConf().setAppName("PrintRecMsg").setMaster("local[2]"); final JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(10000)); String zkQuorum ="host1:2181,host2:2181,host3:2181"; String group = "mygroup"; JavaPairReceiverInputDStream<String, String> lines = KafkaUtils.createStream(jssc, zkQuorum, group, topicmap); JavaDStream<String> msg = lines.map(new Function<Tuple2<String,String>, String>() { @Override public String call(Tuple2<String, String> tuple2) throws Exception { return tuple2._1() + "," + tuple2._2(); } }); msg.print(20); jssc.start(); jssc.awaitTermination(); } }

 

 
（4）分析：
创建的流只需要三个和Kafka有关的参数：zk集群地址，消费者组，topicMap。
createStream是走的Zookeeper去获取对应集群broker信息，然后进行消费。
 
【实验2】无法运行
（1）kafka环境：kafka+Kerberos+Sentry认证
（2）使用jar包：（同上）
（3）核心代码：由于需要添加额外的kafka参数，因此采用了另一个“createStream”的重载方法。 
public static void main(String[] args) { SparkConf sparkConf = new SparkConf().setAppName("PrintDirectMsg").setMaster("local[2]"); final JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(10000)); Map<String, Integer> topicmap = new HashMap<>(); topicmap.put("gaoweiurl", 2); HashMap<String, String> kafkaParams = new HashMap<String, String>(); kafkaParams.put("zookeeper.connect", "host1:2181,host2:2181,host3:2181";); kafkaParams.put("group.id", "mygroup"); kafkaParams.put("auto.offset.reset", "largest"); /** 以下是和Kerberos+sentry认证相关 **/ kafkaParams.put("security.protocol", "SASL_PLAINTEXT"); kafkaParams.put("sasl.mechanism", "GSSAPI"); kafkaParams.put("sasl.kerberos.service.name", "kafka"); System.setProperty("java.security.auth.login.config", "/xx/xx/kafka-jaas.conf"); System.setProperty("java.security.krb5.conf", "/xx/xx/krb5.conf"); JavaPairInputDStream<String, String> lines = KafkaUtils.createStream(jssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topicmap, StorageLevel.MEMORY_AND_DISK_SER_2()); JavaDStream<String> msg = lines.map(new Function<Tuple2<String, String>, String>() { @Override public String call(Tuple2<String, String> tuple2) throws Exception { return tuple2._1() + "," + tuple2._2(); } }); msg.print(20); jssc.start(); jssc.awaitTermination(); } 
 （4）结果及分析：
（4）问题及分析
 
 
 
Zookeeper可以正常连接，而且日志显示，已经成功通过Kerberos+sentry认证。
但是在开始消费消息的时候，一直报一个错误：
通过异常堆栈一行行的查找源码，首先：
==>“at org.apache.kafka.common.utils.Utils.formatAddress(Utils.java:312)”
抛出了空指针异常。进入代码发现：
说明传入host为空。
==>“at kafka.cluster.Broker.connectionString(Broker.scala:62)”
说明是创建Broker对象的时候，调用“connectionString(host,port)”出现的。
查看这个方法，是Broker的一个成员方法
创建Broker的时候，入参host为空，那么是谁创建Broker对象呢？
==>“at kafka.client.ClientUtils$$anonfun$fetchTopicMetadata$5.apply(ClientUtils.scala:89)”
 
这行代码说明，是有人先创建了“brokers”，然后在执行“fetchTopicMetadata”方法的时候，
执行broker.map(_.connectionString)，而brokers的“host”为空，所以空指针异常。
这里的结果已经明确了，但是，还没有找到brokers对象创建的地方。
异常堆栈在这个地方的时候就断了。剩下的就是找到brokers的创建位置。
接下来从下往上看异常：
==>“at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)”
线程“LeaderFinderThread”启动之后，开始执行doWork()方法。
然后创建了brokers，然后通过“ClientUtlis.fetchTopicMetadata”触发了前面的空指针错误。
那么“getAllBrokersInCluster(zkClient)”方法是怎么生成brokers的呢？
首先，这个例子是通过Zookeeper获取broker的，这个地方基本上可以确定是通过zk获取kafka的broker信息。
通过
进入“getAllBrokersInCluster”方法：
其中，BrokerIdspath为：
接下来，
总之，就是会读取“brokers/ids/id”中的信息，返回brokerInfo，然后创建broker。
回到Broker，进入“createBroker”方法查看：
最终，看看配置了Kerberos+Sentry的kafka的broker信息吧，
呵呵，比较下不加Kerberos+sentry的kafka的broker信息：
呵呵。
 
【实验3】无法运行
（1）kafka环境：kafka+Kerberos+Sentry认证
（2）使用jar包：（同上）
（3）核心代码：使用createDirectStream，不用Zookeeper，直接连接broker。 
import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.Function; import org.apache.spark.streaming.Duration; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import org.apache.spark.streaming.kafka.KafkaUtils; import kafka.serializer.StringDecoder; import scala.Tuple2; public class PrintDirectMsgDirect { public static void main(String[] args) { SparkConf sparkConf = new SparkConf().setAppName("PrintDirectMsg").setMaster("local[2]"); final JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(10000)); int numThreads = 2; Map<String, Integer> topicmap = new HashMap<>(); topicmap.put("gaoweiurl", numThreads); Set<String> topicSet = new HashSet<>(); topicSet.add("gaoweiurl"); HashMap<String, String> kafkaParams = new HashMap<String, String>(); kafkaParams.put("bootstrap.servers", "node86:9092,node99:9092,node101:9092"); kafkaParams.put("metadata.broker.list", "node86:9092,node99:9092,node101:9092"); kafkaParams.put("group.id", "spark-executor-kafka_shjs_wlpt"); kafkaParams.put("auto.offset.reset", "smallest"); kafkaParams.put("enable.auto.commit", "true"); kafkaParams.put("security.protocol", "SASL_PLAINTEXT"); kafkaParams.put("sasl.mechanism", "GSSAPI"); kafkaParams.put("sasl.kerberos.service.name", "kafka"); System.setProperty("java.security.auth.login.config", "D:\\Kerberos\\kafka-jaas.conf"); System.setProperty("java.security.krb5.conf", "D:\\Kerberos\\krb5.conf"); JavaPairInputDStream<String, String> lines= KafkaUtils.createDirectStream(jssc,  String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topicSet); JavaDStream<String> msg = lines.map(new Function<Tuple2<String, String>, String>() { @Override public String call(Tuple2<String, String> tuple2) throws Exception { return tuple2._1() + "," + tuple2._2(); } }); msg.print(20); jssc.start(); jssc.awaitTermination(); } }
（4）结果及分析
然后，百度了下，呵呵。
Spark的1.x版本不支持“Secure Kafka”。
【实验4】失败后成功
换了高版本的spark和sparkStreaming和sparkStreamingKafka的jar包
（1）kafka环境：kafka+Kerberos+Sentry认证
（2）使用jar包：全部是高版本 
 <spark.version>2.1.0</spark.version>
        <spark-streaming-kafka.version>2.1.0</spark-streaming-kafka.version> 
<!-- spark core 核心依赖包 -->
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.10</artifactId>
                <version>${spark.version}</version>
            </dependency>

            <!-- spark streaming 依赖包 -->
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.10</artifactId>
                <version>${spark.version}</version>
            </dependency>

            <!-- spark streaming kafka 依赖包 -->
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming-kafka-0-10_2.10</artifactId>
                <version>${spark-streaming-kafka.version}</version>
            </dependency>
（3）核心代码： 
package com.ustcinfo.ishare.bdp.spark import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent /** * @Description: 所有spark任务的总入口 * @author: Beethoven.S * @date: 2017/9/14 13:50 * @e-mail: sheng.gang@ustcinfo.com */ object sparkJobExecutor { /** krb5.conf配置文件 **/ val KRB5_CONF: String = "D:\\Kerberos\\krb5.conf"
    /** JAAS配置文件 **/ val KAFKA_JAAS_CONF: String = "D:\\Kerberos\\kafka-jaas.conf"
    /** kafka broker地址，多个broker用逗号分开 **/ val KAFKA_BROKERS: String = "node86:9092,node99:9092,node101:9092" def main(args: Array[String]) { /** 添加Kerberos认证所需的JAAS配置文件到运行时环境 **/ System.setProperty("java.security.auth.login.config", KAFKA_JAAS_CONF) /** 添加krb5配置文件到运行时环境 **/ System.setProperty("java.security.krb5.conf", KRB5_CONF) val sparkConf = new SparkConf().setMaster("local[4]").setAppName("sparkStreaming") val streamingContext = new StreamingContext(sparkConf, Seconds(10)) val stream = KafkaUtils.createDirectStream[String, String]( streamingContext, PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array("gaoweiurl"), Map( "bootstrap.servers" -> "node86:9092,node99:9092,node101:9092", "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer", "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer", "group.id" -> "kafka_shjs_wlpt", "enable.auto.commit" -> "true", "auto.offset.reset" -> "earliest", "sasl.kerberos.service.name" -> "kafka", "security.protocol" -> "SASL_PLAINTEXT" )) ) stream.foreachRDD(kv => { println("============> ") kv.foreach(x => println("RDD==> " + x)) }) streamingContext.start() streamingContext.awaitTermination() } }
 
然后，在没有数据写入的时候，很正常，但是一旦开始接收数据之后，就会出现如下错误：
没错，sparkStreaming开始监听并且连接的时候，用的消费者组ID确实是我代码中配置的：
但是，一旦开始接收消息，通过RDD读取数据的时候，groupId居然被自动添加了“spark-executor-”的前缀。
（我这个开始设置的“spark-executor-kafka_shjs_wlpt”这个就是看看是不是有了前缀就不再加的，结果，
还是自动的添加了前缀，成了“spark-executor-spark-executor-kafka_shjs_wlpt”）。
然后，在spark运行的时候发现了这句话：
 
找到“KafkaUtils”代码中，追踪到万恶之源：
而且是，只要收到消息，创建RDD的时候会这么干：
 
解决方式就是覆盖这个源码：
 
注意覆盖的时候，包的名称路径必须要和源码路径一模一样，否则会出现scala的私有依赖引用问题。
然后再次，执行：
kafka的数据可以正常读取。
 
 


免责声明！

本站转载的文章为个人学习借鉴使用，本站对版权不负任何法律责任。如果侵犯了您的隐私权益，请联系本站邮箱yoyou2525@163.com删除。



猜您在找 Sparkstreaming and Kafka Flume+Kafka+Sparkstreaming日志分析 SparkStreaming操作Kafka SparkStreaming消费Kafka数据限速问题 Kafka集成SparkStreaming SparkStreaming 整合kafka Demo SparkStreaming-Kafka集成 SparkStreaming+Kafka整合 Scala创建SparkStreaming获取Kafka数据代码过程 SparkStreaming以Direct的方式对接Kafka
 
粤ICP备18138465号  © 2018-2025 CODEPRJ.COM