spring-sparkstreaming-kafka10集成实现和疑难杂症解决

港控/mmm° 2022-06-12 00:21 104阅读 0赞

# 一.前期准备 #

#### 1.开发环境 ####

window7

eclipse

jdk1.8

#### 2.linux环境 ####

zookeeper-3.4.8

hadoop-2.6.4

spark-1.6.0

scala-2.10.6

kafka\_2.10-0.10.1.0

各环境的安装和部署请自行准备。

# 二.疑难杂症 #

#### 1. spark+scala+kafka版本要一致 ####

#### 2. org.apache.spark.SparkException: A master URL must be set in your configuration ####

![Center][]

![Image 1][]

原因：

![Center 1][]

![Image 1][]

SparkConf中未设置master。由于Master为cluster程序管理中心，负责接收Client提交的作业，管理Worker，并命令Worker启动Driver和Executor。

解决：

在SparkConf中设置master，本处设置为“local\[2\]”.

local(default)在本地而非集群跑Spark作业，并且只有一个worker thread（所以，并事实上没有并行）  
local\[k\] 在本地跑Spark Application，有k个worker thread  
spark://HOST:PORT连接到指定URL的standalone集群  
mesos://HOST:PORT连接到指定的Mesos集群  
yarn连接到默认的YARN集群。yarn集群在SPARK\_HOME/conf/yarn-site.xml中指定

#### 3. object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord) ####

![Center 2][]

![Image 1][]

原因：

ConsumerRecord是Kafka10版本自带的类。该类为接受kafka消息的key-value类，而异常时该类未序列化。而程序中调用JavaRdd的collect方法，而该方法需要将数据加载到内存，需要进行序列化。

![Center 3][]

![Image 1][]

解决：

方法1：将获取数据集合的形式转化单个数据获取。

![Center 4][]

![Image 1][]

方法2：通过kryo序列化ConsumerRecord类

![Center 5][]![Center 6][]

![Image 1][]

# 三.代码实现 #

#### 1. sparkStreaming+kafka处理 ####

package com.lm.spark;
    
    import java.io.Serializable;
    import java.util.Arrays;
    import java.util.Collection;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.function.VoidFunction;
    import org.apache.spark.api.java.function.VoidFunction2;
    import org.apache.spark.streaming.Durations;
    import org.apache.spark.streaming.Time;
    import org.apache.spark.streaming.api.java.JavaInputDStream;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    import org.apache.spark.streaming.kafka010.ConsumerStrategies;
    import org.apache.spark.streaming.kafka010.KafkaUtils;
    import org.apache.spark.streaming.kafka010.LocationStrategies;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    @Component
    public class SparkStreamingKafka implements Serializable {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 1L;
    
    	public static Logger LOGGER = LoggerFactory.getLogger(SparkStreamingKafka.class);
    
    	@Value("${spark.appname}")
    	private String appName;
    	@Value("${spark.master}")
    	private String master;
    	@Value("${spark.seconds}")
    	private long second;
    	@Value("${kafka.metadata.broker.list}")
    	private String metadataBrokerList;
    	@Value("${kafka.auto.offset.reset}")
    	private String autoOffsetReset;
    	@Value("${kafka.topics}")
    	private String kafkaTopics;
    	@Value("${kafka.group.id}")
    	private String kafkaGroupId;
    
    	public void processSparkStreaming() throws InterruptedException {
    		// 1.配置sparkconf,必须要配置master
    		SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
    		conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
    		conf.set("spark.kryo.registrator", "com.lm.kryo.MyRegistrator");
    
    		// 2.根据sparkconf 创建JavaStreamingContext
    		JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(second));
    
    		// 3.配置kafka
    		Map<String, Object> kafkaParams = new HashMap<>();
    		kafkaParams.put("bootstrap.servers", metadataBrokerList);
    		kafkaParams.put("key.deserializer", StringDeserializer.class);
    		kafkaParams.put("value.deserializer", StringDeserializer.class);
    		kafkaParams.put("group.id", kafkaGroupId);
    		kafkaParams.put("auto.offset.reset", autoOffsetReset);
    		kafkaParams.put("enable.auto.commit", false);
    
    		// 4.kafka主题
    		Collection<String> topics = Arrays.asList(kafkaTopics.split(","));
    
    		// 5.创建SparkStreaming输入数据来源input Stream
    		final JavaInputDStream<ConsumerRecord<String, String>> stream =
    				KafkaUtils.createDirectStream(jsc, LocationStrategies.PreferConsistent(),
    						ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams));
    
    		// 6.spark rdd转化和行动处理
    		stream.foreachRDD(new VoidFunction2<JavaRDD<ConsumerRecord<String, String>>, Time>() {
    
    			private static final long serialVersionUID = 1L;
    
    			@Override
    			public void call(JavaRDD<ConsumerRecord<String, String>> v1, Time v2) throws Exception {
    
    				List<ConsumerRecord<String, String>> consumerRecords = v1.collect();
    
    				System.out.println("获取消息:" + consumerRecords.size());
    
    			}
    		});
    
    		// 6. 启动执行
    		jsc.start();
    		// 7. 等待执行停止，如有异常直接抛出并关闭
    		jsc.awaitTermination();
    	}
    }

#### 2. kryo序列化接收器 ####

package com.lm.kryo;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.spark.serializer.KryoRegistrator;
    
    import com.esotericsoftware.kryo.Kryo;
    
    public class MyRegistrator implements KryoRegistrator {
    
    	@Override
    	public void registerClasses(Kryo arg0) {
    		arg0.register(ConsumerRecord.class);
    	}
    
    }

# 四.代码地址 #

github:[https://github.com/a123demi/spring-sparkstreaming-kafka-10][https_github.com_a123demi_spring-sparkstreaming-kafka-10]

oschina:[http://git.oschina.net/a123demi/spring-sparkstreaming-kafka-10][http_git.oschina.net_a123demi_spring-sparkstreaming-kafka-10]

[Center]: /images/20220612/7e0a21ec8e2c41dc95149ff76a3057f6.png
[Image 1]: 
[Center 1]: /images/20220612/60d950ed2ab541ea9cb3dd2e870aeab3.png
[Center 2]: /images/20220612/e32937c8842846c699754993d5c400a2.png
[Center 3]: /images/20220612/b13d9087d0154c8ab9e8bd0cf8df00fc.png
[Center 4]: /images/20220612/c9892493228048cfabdd2faff0ccc8e3.png
[Center 5]: https://img-blog.csdn.net/20170710194558780?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvYTEyM2RlbWk=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 6]: /images/20220612/aec8d49e4095424e91677449588ca15f.png
[https_github.com_a123demi_spring-sparkstreaming-kafka-10]: https://github.com/a123demi/spring-sparkstreaming-kafka-10
[http_git.oschina.net_a123demi_spring-sparkstreaming-kafka-10]: http://git.oschina.net/a123demi/spring-sparkstreaming-kafka-10