Insert rows with null values to HBase with Spark

Question

Trying to write a Spark dataframe with null column values, I am getting exception about unsupported data type. There is a sample of code:

val sql = spark.sqlContext 

case class Person(name: String, email: String, height: Float) 

var personDS = Seq(Person("alice", "alice@alice.com", 4.5f),Person("bob", null, 5.1f) ).toDS 
//                                                                        ----
//                                                                          ^ null value

personDS.write.format("org.apache.hadoop.hbase.spark").option("hbase.columns.mapping", "name STRING :key, email STRING c:email, height FLOAT p:height").option("hbase.table", "test").option("hbase.spark.use.hbasecontext", false).option("spark.hadoop.validateOutputSpecs", false).save()

The exception is

org.apache.spark.SparkException: Task failed while writing rows
        at org.apache.spark.internal.io.SparkHadoopWriter$.org$apache$spark$internal$io$SparkHadoopWriter$$executeTask(SparkHadoopWriter.scala:155)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$3.apply(SparkHadoopWriter.scala:83)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$3.apply(SparkHadoopWriter.scala:78)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.Exception: unsupported data type StringType
        at org.apache.spark.sql.datasources.hbase.Utils$.toBytes(Utils.scala:87)
        at org.apache.hadoop.hbase.spark.HBaseRelation$$anonfun$org$apache$hadoop$hbase$spark$HBaseRelation$$convertToPut$1$2.apply(DefaultSource.scala:225)
        at org.apache.hadoop.hbase.spark.HBaseRelation$$anonfun$org$apache$hadoop$hbase$spark$HBaseRelation$$convertToPut$1$2.apply(DefaultSource.scala:224)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
        at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
        at org.apache.hadoop.hbase.spark.HBaseRelation.org$apache$hadoop$hbase$spark$HBaseRelation$$convertToPut$1(DefaultSource.scala:224)
        at org.apache.hadoop.hbase.spark.HBaseRelation$$anonfun$insert$1.apply(DefaultSource.scala:231)
        at org.apache.hadoop.hbase.spark.HBaseRelation$$anonfun$insert$1.apply(DefaultSource.scala:231)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:129)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:127)
        at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1442)
        at org.apache.spark.internal.io.SparkHadoopWriter$.org$apache$spark$internal$io$SparkHadoopWriter$$executeTask(SparkHadoopWriter.scala:139)
        ... 10 more

Of course, I can separate Persons with null values like the following:

case class Person(name: String, email: String, height: Float) 
case class Person_without_email(name: String, height: Float) 

var personDS = Seq(Person("alice", "alice@alice.com", 4.5f)).toDS 
var person_without_email_DS = Seq(Person(Person_without_email("bob", 5.1f) )).toDS 

personDS.write.format("org.apache.hadoop.hbase.spark").option("hbase.columns.mapping", "name STRING :key, email STRING c:email, height FLOAT p:height").option("hbase.table", "test").option("hbase.spark.use.hbasecontext", false).option("spark.hadoop.validateOutputSpecs", false).save()    
person_without_email_DS.write.format("org.apache.hadoop.hbase.spark").option("hbase.columns.mapping", "name STRING :key, height FLOAT p:height").option("hbase.table", "test").option("hbase.spark.use.hbasecontext", false).option("spark.hadoop.validateOutputSpecs", false).save()

This approach looks ugly. I should split original dataframe on large number of parts (~ !(number of fields)). Is there more convenient way to process rows with null values?

HBase 2.1.0-cdh6.2.1
Spark 2.4.0-cdh6.2.1

Insert rows with null values to HBase with Spark

0 Answers