SHC with Spark Structured Streaming

[EDALJO]

Hi,

I have a Spark Structured Streaming application where I'd like to write streaming data to HBase using SHC. It reads data from a location where new csv files continuously are being created. The defined catalog works for writing a DataFrame with identical data into HBase.
The key components of my streaming application are a DataStreamReader and a DataStreamWriter.

val inputDataStream = spark
      .readStream
      .option("sep", ",")
      .schema(schema)
      .csv("/path/to/data/*.csv")

inputDataStream
      .writeStream
      .outputMode("append")
      .options(
        Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "2"))
      .format("org.apache.spark.sql.execution.datasources.hbase")
      .start

When running the application I'm getting the following message:

Exception in thread "main" java.lang.UnsupportedOperationException: Data source org.apache.spark.sql.execution.datasources.hbase does not support streamed writing at org.apache.spark.sql.execution.datasources.DataSource.createSink(DataSource.scala:285) at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:286) at my.package.SHCStreamingApplication$.main(SHCStreamingApplication.scala:153) at my.package.SHCStreamingApplication.main(SHCStreamingApplication.scala) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:755) at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119) at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Does anyone know a solution or way/workaround to still use the SHC for writing structured streaming data to HBase?
Thanks in advance!

[sutugin]

You can write your custom sink provider, inherited from StreamSinkProvider, this is my implementation:

package HBase
import org.apache.spark.internal.Logging
import org.apache.spark.sql.execution.streaming.Sink
import org.apache.spark.sql.sources.{DataSourceRegister, StreamSinkProvider}
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.sql.execution.datasources.hbase._

class HBaseSink(options: Map[String, String]) extends Sink with Logging {
  // String with HBaseTableCatalog.tableCatalog
  private val hBaseCatalog = options.get("hbasecat").map(_.toString).getOrElse("")

  override def addBatch(batchId: Long, data: DataFrame): Unit = synchronized {   
    val df = data.sparkSession.createDataFrame(data.rdd, data.schema)
    df.write
      .options(Map(HBaseTableCatalog.tableCatalog->hBaseCatalog,
        HBaseTableCatalog.newTable -> "5"))
      .format("org.apache.spark.sql.execution.datasources.hbase").save()
  }
}

class HBaseSinkProvider extends StreamSinkProvider with DataSourceRegister {
  def createSink(
                  sqlContext: SQLContext,
                  parameters: Map[String, String],
                  partitionColumns: Seq[String],
                  outputMode: OutputMode): Sink = {
    new HBaseSink(parameters)
  }

  def shortName(): String = "hbase"
}

This is example, how to use ():

inputDF.
   writeStream.
   queryName("hbase writer").
   format("HBase.HBaseSinkProvider").
   option("checkpointLocation", checkPointProdPath).
   option("hbasecat", catalog).
   outputMode(OutputMode.Update()).
   trigger(Trigger.ProcessingTime(30.seconds)).
   start

[EDALJO]

Thanks for your answer - exactly the type of solution I was looking for. I only had time to test it quickly, but seems to be working perfectly!

[sutugin]

Excellent, glad to help!!!

[hamroune]

Thank you, much helps

[merfill]

I've implemented your solution with HBaseSinkProvider by following steps:

1. Clone shc.
2. Put HBaseSinkProvider.scala to core/src/main/scala/org/apache/spark/sql/execution/datasources/hbase.
3. Compile shc.
4. Run spark-submit with --jars some_path/shc/core/target/shc-core-1.1.2-2.2-s_2.11-SNAPSHOT.jar

My code is written in python, I'm including it below. The error is: pyspark.sql.utils.StreamingQueryException: u'Queries with streaming sources must be executed with writeStream.start();;
I'm new in Spark and haven't experience Scala, so I cannot understand the problem. Can you please help me with it?

def consume(schema_name, brokers, topic, group_id):
spark = SparkSession
.builder
.appName('SparkConsumer')
.config('hbase.zookeeper.property.clientPort', '2282')
.getOrCreate()

print 'read Avro schema from file: {}...'.format(schema_name)
schema = avro.schema.parse(open(schema_name, 'rb').read())
reader = avro.io.DatumReader(schema)
print 'the schema is read'

rows = spark \
    .readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', brokers) \
    .option('subscribe', topic) \
    .option('group.id', group_id) \
    .option('maxOffsetsPerTrigger', 1000) \
    .option("startingOffsets", "earliest") \
    .load()
rows.printSchema()

schema = StructType([ \
        StructField('consumer_id', StringType(), False), \
        StructField('audit_system_id', StringType(), False), \
        StructField('object_path', StringType(), True), \
        StructField('object_type', StringType(), False), \
        StructField('what_action', StringType(), False), \
        StructField('when', LongType(), False), \
        StructField('where', StringType(), False), \
        StructField('who', StringType(), True), \
        StructField('workstation', StringType(), True) \
    ])

def decode_avro(msg):
    bytes_reader = io.BytesIO(bytes(msg))
    decoder = avro.io.BinaryDecoder(bytes_reader)
    data = reader.read(decoder)
    return (\
            data['consumer_id'],\
            data['audit_system_id'],\
            data['object_path'],\
            data['object_type'],\
            data['what_action'],\
            data['when'],\
            data['where'],\
            data['who'],\
            data['workstation']\
           )

udf_decode_avro = udf(decode_avro, schema)

values = rows.select('value')
values.printSchema()

changes = values.withColumn('change', udf_decode_avro(col('value'))).select('change.*')
changes.printSchema()

change_catalog = '''
{
    "table":
    {
        "namespace": "default",
        "name": "changes",
        "tableCoder": "PrimitiveType"
    },
    "rowkey": "consumer_id",
    "columns":
    {
        "consumer_id": {"cf": "rowkey", "col": "consumer_id", "type": "string"},
        "audit_system_id": {"cf": "d", "col": "audit_system_id", "type": "string"},
        "object_path": {"cf": "d", "col": "object_path", "type": "string"},
        "object_type": {"cf": "d", "col": "object_type", "type": "string"},
        "what_action": {"cf": "d", "col": "what_action", "type": "string"},
        "when": {"cf": "t", "col": "when", "type": "bigint"},
        "where": {"cf": "d", "col": "where", "type": "string"},
        "who": {"cf": "d", "col": "who", "type": "string"},
        "workstation": {"cf": "d", "col": "workstation", "type": "string"}
    }
}'''

query = changes \
    .writeStream \
    .format('HBase.HBaseSinkProvider')\
    .option('hbasecat', change_catalog) \
    .option("checkpointLocation", '/tmp/checkpoint') \
    .outputMode("append") \
    .start()

query.awaitTermination()

[sutugin]

Try to write to the console orfile, will there be the same error?

[merfill]

No, when I'm trying to write records to console, everything is OK. I'm using following python code instead
HBase.HBaseSinkProvider:

query = changes \
         .writeStream \
        .outputMode("append") \
        .format('console') \
        .start()

Output is somethinf like this:

Batch: 7

+-----------+---------------+--------------------+-----------+-----------+-------------+-------------+----------------+--------------------+
|consumer_id|audit_system_id| object_path|object_type|what_action| when| where| who| workstation|
+-----------+---------------+--------------------+-----------+-----------+-------------+-------------+----------------+--------------------+
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Removed|1520170874201|172.28.26.190| n/a| n/a|
| 111| 222|\172.28.26.190\T...| File| Added|1520171584122|172.28.26.190|PD6\fsatestuser1|win10test1.pd6.local|
| 111| 222|\172.28.26.190\T...| File| Added|1520171584126|172.28.26.190|PD6\fsatestuser1|win10test1.pd6.local|
+-----------+---------------+--------------------+-----------+-----------+-------------+-------------+----------------+--------------------+
only showing top 20 rows

2018-06-27 14:01:26 INFO WriteToDataSourceV2Exec:54 - Data source writer org.apache.spark.sql.execution.streaming.sources.MicroBatchWriter@30f35fc5 committed.
2018-06-27 14:01:26 INFO SparkContext:54 - Starting job: start at NativeMethodAccessorImpl.java:0
2018-06-27 14:01:26 INFO DAGScheduler:54 - Job 15 finished: start at NativeMethodAccessorImpl.java:0, took 0,000037 s
2018-06-27 14:01:26 INFO MicroBatchExecution:54 - Streaming query made progress: {
"id" : "1e6076ad-b403-46ca-9438-e4913660700d",
"runId" : "7740c36b-dc2d-4880-8ea6-0efde89b1ef5",
"name" : null,
"timestamp" : "2018-06-27T11:01:26.389Z",
"batchId" : 7,
"numInputRows" : 669,
"inputRowsPerSecond" : 883.7516512549538,
"processedRowsPerSecond" : 1327.3809523809523,
"durationMs" : {
"addBatch" : 471,
"getBatch" : 4,
"getOffset" : 2,
"queryPlanning" : 11,
"triggerExecution" : 503,
"walCommit" : 15
},
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[changes]]",
"startOffset" : {
"changes" : {
"0" : 7000
}
},
"endOffset" : {
"changes" : {
"0" : 7669
}
},
"numInputRows" : 669,
"inputRowsPerSecond" : 883.7516512549538,
"processedRowsPerSecond" : 1327.3809523809523
} ],
"sink" : {
"description" : "org.apache.spark.sql.execution.streaming.ConsoleSinkProvider@3715c84c"
}
}

[sutugin]

Unfortunately I am now without a computer, try to run the outputMode to update, if it does not help and will not be able to find a solution, then email me after July 5, I will try to help.

[merfill]

Unfortunatelly, "update" mode also not work. I've received the same error (see below). Thank you in advance.

pyspark.sql.utils.StreamingQueryException: u'Queries with streaming sources must be executed with writeStream.start();;\nLogicalRDD [key#52, value#53, topic#54, partition#55, offset#56L, timestamp#57, timestampType#58], true\n\n=== Streaming Query ===\nIdentifier: [id = 66509eea-e706-4745-9b35-f82f09752b43, runId = c0e78fe7-f978-4f64-88cd-801d766f78c5]\nCurrent Committed Offsets: {}\nCurrent Available Offsets: {KafkaSource[Subscribe[changes]]: {"changes":{"0":7669}}}\n\nCurrent State: ACTIVE\nThread State: RUNNABLE\n\nLogical Plan:\nProject [change#22.consumer_id AS consumer_id#25, change#22.audit_system_id AS audit_system_id#26, change#22.object_path AS object_path#27, change#22.object_type AS object_type#28, change#22.what_action AS what_action#29, change#22.when AS when#30L, change#22.where AS where#31, change#22.who AS who#32, change#22.workstation AS workstation#33]\n+- Project [value#8, decode_avro(value#8) AS change#22]\n +- Project [value#8]\n +- StreamingExecutionRelation KafkaSource[Subscribe[changes]], [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]\n'

[ddkongbb]

Please try to use this version of shc (https://github.com/sutugin/shc) and compile with corresponding hbase/phoenix version. I have tried to use this without arvo format perfectly.
/**

• In option must be specified string with HBaseTableCatalog.tableCatalog
• {{{
• inputDF.
• writeStream.
• format("hbase").
• option("checkpointLocation", checkPointProdPath).
• options(Map("hbase.catalog->catalog)).
• outputMode(OutputMode.Update()).
• trigger(Trigger.ProcessingTime(30.seconds)).
• start
• }}}
  /
  For using arvo, please try the following.
  /*
• In option must be specified string with HBaseTableCatalog.tableCatalog
• {{{
• inputDF.
• writeStream.
• format("hbase").
• option("checkpointLocation", checkPointProdPath).
• options(Map("hbase.schema_array"->schema_array,"hbase.schema_record"->schema_record, hbase.catalog->catalog)).
• outputMode(OutputMode.Update()).
• trigger(Trigger.ProcessingTime(30.seconds)).
• start
• }}}
  */

[swarup5s]

Hello @sutugin .. I've implemented your solution. However data is not getting updated in HBase. It's not even throwing any exception too. Can you suggest anything in this regard?

[sutugin]

Hi @swarup5s ,if you give me the implementation code and how you use it, show me the logs, maybe we can find the problem together.

[swarup5s]

Hi @sutugin thanks for your help. Appreciate it.

//this class is under ...org/apache/spark/sql/execution/streaming/
//path. I'm executing from IntelliJ..
package org.apache.spark.sql.execution.streaming

import org.apache.spark.sql.execution.datasources.hbase.Logging
import org.apache.spark.sql.sources.{DataSourceRegister, StreamSinkProvider}
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, Row, SQLContext}

import org.apache.spark.sql.execution.datasources.hbase._

class HBaseSink(options: Map[String, String]) extends Sink with Logging {
// String with HBaseTableCatalog.tableCatalog
private val hBaseCatalog = options.get("hbasecat").map(_.toString).getOrElse("")

override def addBatch(batchId: Long, data: DataFrame): Unit = synchronized {
val df = data.sparkSession.createDataFrame(data.rdd, data.schema)
df.write
.options(Map(HBaseTableCatalog.tableCatalog->hBaseCatalog))
.format("org.apache.spark.sql.execution.datasources.hbase").save()
}
}

class HBaseSinkProvider extends StreamSinkProvider with DataSourceRegister {
def createSink(
sqlContext: SQLContext,
parameters: Map[String, String],
partitionColumns: Seq[String],
outputMode: OutputMode): Sink = {
new HBaseSink(parameters)
}

def shortName(): String = "hbase"
}

/**
*My code goes here------------------------------------------------------------------------------
*/
//...
//...

def catalog = s"""{
|"table":{"namespace":"default", "name":"hbase_table"},
|"rowkey":"key",
|"columns":{
|"rowkey":{"cf":"rowkey", "col":"key", "type":"string"},
|"subscriberName":{"cf":"subscriberInfo", "col":"Name", "type":"string"},
|"subscriberNumber":{"cf":"subscriberInfo", "col":"PhoneNumber", "type":"string"},
|"messageTemplate":{"cf":"messageInfo", "col":"template", "type":"string"},
|"lastTS":{"cf":"messageInfo", "col":"ts", "type":"String"}
|}
|}""".stripMargin

def withCatalog(cat: String): DataFrame = {
spark.sqlContext
.read
.options(Map(HBaseTableCatalog.tableCatalog->cat))
.format("org.apache.spark.sql.execution.datasources.hbase")
.load()
}
val BaseRecorddf = withCatalog(catalog) //record from HBase as Batch or normal Dataframe

val streamdf = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "server_addr")
.option("subscribe", "topic_name")
.option("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
.option("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
.option("auto.offset.reset","earliest")
.option("group.id","realTimeStream")
.option("enable.auto.commit",true: java.lang.Boolean)
.load()

//streaming Dataframe here
val schemaStreamDf = streamdf
  .selectExpr("CAST(key AS STRING) as IMSI", "CAST(value AS STRING) as Loc")

//...

//some biz logic here and some join between the batch and streaming dataframe and finaldf is the streaming DF
//which should be written back to the HBase in real time.

try{

    finaldfd.
      writeStream.
      format("org.apache.spark.sql.execution.streaming.HBaseSinkProvider").
      option("checkpointLocation", "some_path_here").
      option("hbasecat",catalog).
      outputMode(OutputMode.Update()).
      trigger(Trigger.ProcessingTime(some_seconds))
      .start()
      .awaitTermination()
    
  }
catch {
  case e: Exception => println(e)
}

I've done some changes. now some StreamingQueryException is thrown but not sure what is getting wrong being a novice. Here's the logs:

org.apache.spark.sql.streaming.StreamingQueryException: null
=== Streaming Query ===
Identifier: [id = a0997feb-efeb-4976-b8eb-5efaa3c1b8c9, runId = 6f4181f0-ffd4-4aa9-a0bc-475457e4e93e]
Current Committed Offsets: {}
Current Available Offsets: {KafkaSource[Subscribe[topic_name]]: {"topic_name":{"2":282,"1":305,"0":293}}}

Current State: ACTIVE
Thread State: RUNNABLE

Logical Plan:
Project [rowkey#75, subscriberName#1, subscriberNumber#2, messageTemplate#3, 1537292624 AS lastTS#82]
+- Project [rowkey#11 AS rowkey#75, subscriberName#1, subscriberNumber#2, messageTemplate#3, lastTS#4]
+- Filter (cast(cast(lastTS#4 as bigint) as double) < 1.536990223E9)
+- Project [rowkey#11, subscriberName#1, subscriberNumber#2, messageTemplate#3, lastTS#4]
+- Project [rowkey#11, subscriberName#1, subscriberNumber#2, messageTemplate#3, lastTS#4, Loc#46]
+- Join Inner, (rowkey#11 = rowkey#45)
:- Project [rowkey#0 AS rowkey#11, subscriberName#1, subscriberNumber#2, messageTemplate#3, lastTS#4]
: +- Relation[rowkey#0,subscriberName#1,subscriberNumber#2,messageTemplate#3,lastTS#4] HBaseRelation(Map(catalog -> {
"table":{"namespace":"default", "name":"hbase_table"},
"rowkey":"key",
"columns":{
"rowkey":{"cf":"rowkey", "col":"key", "type":"string"},
"subscriberName":{"cf":"subscriberInfo", "col":"Name", "type":"string"},
"subscriberNumber":{"cf":"subscriberInfo", "col":"PhoneNumber", "type":"string"},
"messageTemplate":{"cf":"messageInfo", "col":"template", "type":"string"},
"lastTS":{"cf":"messageInfo", "col":"ts", "type":"String"}
}
}),None)
+- Project [cast(key#30 as string) AS rowkey#45, cast(value#31 as string) AS Loc#46]
+- StreamingExecutionRelation KafkaSource[Subscribe[topic_name]], [key#30, value#31, topic#32, partition#33, offset#34L, timestamp#35, timestampType#36]

On the other hand message is successfully written to the console.

[stefcorda]

Hello @sutugin, first of all thank your your great help. I'm experiencing @swarup5s 's problem when i call:

data.sparkSession.createDataFrame(data.rdd, data.schema)

The same happens whenever i call something like
data.rdd

I think the problem is outside your code, somewhere else. Maybe more spark related?