sparkStreaming中用sparksql时遇到的问题总结

问题一

今天准备用sparkStreaming接入kafka再写入hive，准备在流里面执行sparksql，按照官网的写法，一开始的代码是这样的：

SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("test");
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
SparkSession ss = SparkSession.builder()
                .config(sparkConf)
                .enableHiveSupport()
                .getOrCreate();
JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, Durations.seconds(20));

然后执行sql

ss.sql("show databases");

咋一看没啥毛病，执行后发现报错了。。。。报错如下：

百度后告诉我，因为创建了两个sparkcontext
请教同事，告诉我把SparkSession 和JavaStreamingContext 的代码位置换一下。果然执行后不报错了。
然而。。。。
发现个新问题
不管我在hive中创建多少个database，执行后都只有一个default库。懵逼。。。
继续找问题，发现。。。。
SparkSession.builder()会创建一个sparkContext() ;
new JavaStreamingContext(sparkConf, Durations.seconds(20)) 也创建一个sparkContext；
spark 只能使用一个全局的SparkContext；
所以，SparkSession创建时的sparkConf就不是初始化的sparkConf。
后来代码改成如下：

SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("test");
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
SparkSession ss = SparkSession.builder()
		.config(sparkConf)
		.enableHiveSupport()
		.getOrCreate();
JavaStreamingContext jsc = new JavaStreamingContext(new JavaSparkContext(ss.sparkContext()), Durations.seconds(20));
ss.sql("show databases");

用JavaStreamingContext的另一个构造方法，从 SparkSession 中获取到JavaSparkContext。

问题二

问题：org.apache.hadoop.hive.ql.metadata.Hive.loadDynamicPartitions(org.apache.hadoop.fs.Path, java.lang.String, java.util.Map, boolean, int, boolean, boolean, long, boolean, org.apache.hadoop.hive.ql.io.AcidUtils$Operation)
原因：找到loadDynamicPartitions这个方法，发现是来自下面的依赖

<groupId>org.spark-project.hive</groupId>
<artifactId>hive-metastore</artifactId>
<version>1.2.1.spark2</version>

<groupId>org.spark-project.hive</groupId>
<artifactId>hive-metastore</artifactId>
<version>1.2.1.spark2</version>

这个依赖是从下面这个依赖继承来的

<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.11</artifactId>
<version>2.4.3</version>

报错信息来看，loadDynamicPartitions这个方法需要10个参数，所以把这个自带的依赖排掉，换个依赖，保证loadDynamicPartitions这个方法的参数是10个参数。
最终换到2.3.4版本，用的是org.apache.hive，不是之前的org.spark-project.hive

<dependency>
	<groupId>org.apache.hive</groupId>
	<artifactId>hive-exec</artifactId>
	<version>2.3.4</version>
</dependency>
<dependency>
	<groupId>org.apache.hive</groupId>
	<artifactId>hive-metastore</artifactId>
	<version>2.3.4</version>
</dependency>

问题解决！

问题三

问题：Caused by: java.io.NotSerializableException: Object of org.apache.spark.streaming.kafka010.DirectKafkaInputDStream is being serialized possibly as a part of closure of an RDD operation. This is because the DStream object is being referred to from within the closure. Please rewrite the RDD operation inside this DStream to avoid this. This has been enforced to avoid bloating of Spark tasks with unnecessary objects.
原因：
采用的是spring-sparkstreaming-kafka10，如果是采用官方文档里上述方式手动提交offset，需要把stream对象的属性标记为static或者transient避免序列化，不然可能在任务提交的时候报DirectKafkaInputDStream 无法序列化导致Task not serializable错误。
解决：
定义一个stream private static(transient) JavaInputDStream<ConsumerRecord<String, String>> stream;
然后在初始化他 stream = KafkaParamsConfig.buildKafkaSourceDStream(topic, jsc);

问题四

sparkstreaming的checkpoint的间隔要是批次间隔的倍数

注：后续其他问题再补充。