Spark 序列化错误:当我将 Spark 流数据插入 HBase 时

Question

我对火花在数据格式方面如何与HBase相互作用感到困惑。例如，当我在以下代码片段中省略"错误"行时，它运行良好......但是添加该行，我发现了与序列化问题相关的错误，指出"任务不可序列化"。

如何更改代码？发生错误的原因是什么？

我的代码如下：

// HBase 
    Configuration hconfig = HBaseConfiguration.create();
    hconfig.set("hbase.zookeeper.property.clientPort", "2222");
    hconfig.set("hbase.zookeeper.quorum", "127.0.0.1");  
    hconn = HConnectionManager.createConnection(hconfig);  
    HTable htable = new HTable(hconf, Bytes.toBytes(tableName));       
// KAFKA configuration 
    Set<String> topics = Collections.singleton(topic); 
    Map<String, String> kafkaParams = new HashMap<>();
    kafkaParams.put("metadata.broker.list", "localhost:9092");
    kafkaParams.put("zookeeper.connect", "localhost:2222");
    kafkaParams.put("group.id", "tag_topic_id");  
//Spark Stream  
    JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
                ssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics );  
    JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {   
        @Override
        public String call(Tuple2<String, String> tuple2)  { 
            return tuple2._2();
        }
    });  
    JavaDStream<String> records = lines.flatMap(new FlatMapFunction<String, String>() {  
        @Override
        public Iterator<String> call(String x) throws IOException {   
////////////// Put into HBase : ERROR ///////////////////// 
            String[] data = x.split(","); 
            if (null != data && data.length > 2 ){ 
                SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");   
                String ts = sdf.format(new Date());
                Put put = new Put(Bytes.toBytes(ts)); 
                put.addImmutable(Bytes.toBytes(familyName), Bytes.toBytes("LINEID"), Bytes.toBytes(data[0]));
                put.addImmutable(Bytes.toBytes(familyName), Bytes.toBytes("TAGID"), Bytes.toBytes(data[1]));
                put.addImmutable(Bytes.toBytes(familyName), Bytes.toBytes("VAL"), Bytes.toBytes(data[2]));
                htable.put(put); // ***** ERROR ******** 
                htable.close();  
            }
            return Arrays.asList(COLDELIM.split(x)).iterator(); 
        } 
    }); 
    records.print();
    ssc.start();
    ssc.awaitTermination();

启动应用程序时，遇到以下错误：

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2037)
at org.apache.spark.streaming.dstream.DStream$$anonfun$flatMap$1.apply(DStream.scala:554)
at org.apache.spark.streaming.dstream.DStream$$anonfun$flatMap$1.apply(DStream.scala:554)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.SparkContext.withScope(SparkContext.scala:682)
at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
at org.apache.spark.streaming.dstream.DStream.flatMap(DStream.scala:553)
at org.apache.spark.streaming.api.java.JavaDStreamLike$class.flatMap(JavaDStreamLike.scala:172)
at org.apache.spark.streaming.api.java.AbstractJavaDStreamLike.flatMap(JavaDStreamLike.scala:42) 
Caused by: java.io.NotSerializableException: org.apache.hadoop.hbase.client.HTable
Serialization stack:
- object not serializable (class: org.apache.hadoop.hbase.client.HTable, value: MCSENSOR;hconnection-0x6839203b)

Answer 1

这里有一个序列化调试器的提示

Caused by: java.io.NotSerializableException: org.apache.hadoop.hbase.client.HTable
Serialization stack:
- object not serializable (class: org.apache.hadoop.hbase.client.HTable, value: MCSENSOR;hconnection-0x6839203b)

将以下部分放在您正在使用它的调用方法（闭包）之前的FlatMapFunction内，这应该可以解决问题

Configuration hconfig = HBaseConfiguration.create();
    hconfig.set("hbase.zookeeper.property.clientPort", "2222");
    hconfig.set("hbase.zookeeper.quorum", "127.0.0.1");  
    hconn = HConnectionManager.createConnection(hconfig);  
    HTable htable = new HTable(hconf, Bytes.toBytes(tableName));