title: Spark serialization error: When I insert Spark Stream data into HBase
body: <p>I'm confused about how spark interact with HBase in terms of data format. For instance, when I omitted the 'ERROR' line in the following code snippet, it runs well... but adding the line, I've caught the error saying 'Task not serializable' related to serialization issue. </p>
  
  <p>How do I change the code? 
  What is the reason why the error happens? </p>
  
  <p>My code is following : </p>
  
  <pre><code>// 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&lt;String&gt; topics = Collections.singleton(topic); 
  
      Map&lt;String, String&gt; kafkaParams = new HashMap&lt;&gt;();
      kafkaParams.put("metadata.broker.list", "localhost:9092");
      kafkaParams.put("zookeeper.connect", "localhost:2222");
      kafkaParams.put("group.id", "tag_topic_id");  
  
  //Spark Stream  
      JavaPairInputDStream&lt;String, String&gt; messages = KafkaUtils.createDirectStream(
                  ssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics );  
  
      JavaDStream&lt;String&gt; lines = messages.map(new Function&lt;Tuple2&lt;String, String&gt;, String&gt;() {   
  
          @Override
          public String call(Tuple2&lt;String, String&gt; tuple2)  { 
              return tuple2._2();
          }
      });  
  
      JavaDStream&lt;String&gt; records = lines.flatMap(new FlatMapFunction&lt;String, String&gt;() {  
  
          @Override
          public Iterator&lt;String&gt; call(String x) throws IOException {   
  
  ////////////// Put into HBase : ERROR ///////////////////// 
              String[] data = x.split(","); 
  
              if (null != data &amp;&amp; data.length &gt; 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();
  </code></pre>
  
  <p>When I start my application, I met the following error:</p>
  
  <pre><code>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)
  </code></pre>