需求：spark streaming消費(fèi)kafka锨匆，offset維護(hù)到kafka topic屯仗，開(kāi)啟checkpoint
環(huán)境：Java鱼填、spark2.3擒贸、spark-streaming-kafka-0-10_2. 11
開(kāi)啟checkpoint后程序遇到序列化異常，解決方案如下：
① 程序用到了自定義ConsumerStrategy窗轩，需要添加序列化
② 異步提交offset到kafka夯秃，commitAsync方法的回調(diào)函數(shù)不能用lambda表達(dá)式。需要實(shí)現(xiàn)OffsetCommitCallback回調(diào)函數(shù)接口并序列化
import com.lazyge.sprk.stream.util.KfkConsumerStrategy;
import org.apache.kafka.clients.consumer.ConsumerConfig;
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.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
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.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.*;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
public class CkptTestSpark {
public static void main(String[] args)throws InterruptedException {
new CkptTestSpark().start();
}
public void start()throws InterruptedException {
JavaStreamingContext ssc = getContext();
JavaInputDStream> stream = getStream(ssc);
AtomicReference offsetRanges =new AtomicReference<>();
JavaDStream wordDstream = stream.transform((Function2>, Time, JavaRDD>) (rdd, time) -> {
if( rdd.rdd()instanceof HasOffsetRanges ){
offsetRanges.set(((HasOffsetRanges) rdd.rdd()).offsetRanges());
}
JavaRDD lines = rdd.map((Function, String>) (record) -> {
return (String)record.value();
});
JavaRDD words = lines.flatMap((FlatMapFunction) (line) -> {
return Arrays.asList(line.split(" ")).iterator();
});
return words;
});
wordDstream.foreachRDD((VoidFunction2, Time>)(rdd, time) -> {
rdd.collect().forEach(System.out::println);
/**
* 開(kāi)啟checkpoint的時(shí)候，OffsetCommitCallback 需要實(shí)現(xiàn)序列化才能使用
* 否則報(bào)錯(cuò)：NotSerializableException
*/
if (stream.dstream()instanceof CanCommitOffsets) {
((CanCommitOffsets)stream.dstream()).commitAsync(offsetRanges.get()/*, (OffsetCommitCallback)(ofs, ex) -> {
System.err.println("開(kāi)啟checkpoint的時(shí)候仓洼，OffsetCommitCallback 需要實(shí)現(xiàn)序列化才能使用");
}*/);
}
});
ssc.start();
System.out.println("beginninignininiinini");
ssc.awaitTermination();
}
private JavaStreamingContext getContext(){
SparkConf conf =new SparkConf()
.setIfMissing("spark.master","local[*]")
.setIfMissing("spark.app.name","ckptTestSpark");
JavaStreamingContext ssc =new JavaStreamingContext(conf, Durations.seconds(5));
ssc.checkpoint("E:\\tmp\\ckptspark");
return ssc;
}
private JavaInputDStream> getStream(JavaStreamingContext ssc){
String brokers ="192.168.99.128:9092,192.168.99.128:9093,192.168.99.128:9094";
String groupId ="testCkpt";
String topics ="first";
Set topicsSet =new HashSet<>(Arrays.asList(topics.split(",")));
Map kafkaParams =new HashMap<>();
kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
/**
* 開(kāi)啟checkpoint的時(shí)候介陶，ConsumerStrategy 需要實(shí)現(xiàn)序列化才能使用
* 否則報(bào)錯(cuò)：NotSerializableException
*/
JavaInputDStream> stream = KafkaUtils.createDirectStream(ssc,LocationStrategies.PreferConsistent(),
new KfkConsumerStrategy(kafkaParams, Arrays.asList(topics)));
/*JavaInputDStream> stream = KafkaUtils.createDirectStream(ssc,LocationStrategies.PreferConsistent(),ConsumerStrategies.Subscribe(topicsSet, kafkaParams));*/
return stream;
}
}
總結(jié)：用scala不會(huì)有那么多問(wèn)題
