wchevreuil commented on code in PR #168:
URL: https://github.com/apache/hbase-connectors/pull/168#discussion_r4094119591
##########
spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProviderSuite.scala:
##########
@@ -320,4 +323,90 @@ class HBaseTableProviderSuite extends AnyFunSuite with
BeforeAndAfterAll with Lo
assert(result.count() == 1)
assert(result.first().getAs[String]("name") == "Eve")
}
+
+ // --- Streaming sink tests ---
+
+ val streamTableName = "test_stream_sink"
+
+ val streamCatalog: String = s"""{
+ |"table":{"namespace":"default", "name":"$streamTableName"},
+ |"rowkey":"key",
+ |"columns":{
+ |"key":{"cf":"rowkey", "col":"key", "type":"string"},
+ |"name":{"cf":"$columnFamily", "col":"name", "type":"string"},
+ |"age":{"cf":"$columnFamily", "col":"age", "type":"string"}
+ |}
+ |}""".stripMargin
+
+ private def loadStreamTable() = {
+ spark.read
+ .format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ .option("catalog", streamCatalog)
+ .option(HBaseSparkConf.HBASE_CONFIG_LOCATION, configFile.getAbsolutePath)
+ .load()
+ }
+
+ test("streaming sink writes and reads back") {
+ val ss = spark
+ import ss.implicits._
+
+ TEST_UTIL.createTable(
+ TableName.valueOf(streamTableName), Bytes.toBytes(columnFamily))
+
+ val checkpointDir = Files.createTempDirectory("hbase-stream-ckpt").toFile
+ checkpointDir.deleteOnExit()
+
+ implicit val sqlCtx = spark.sqlContext
+ val source = MemoryStream[(String, String, String)]
+
+ source.addData(
+ ("srow000", "Frank", "32"),
+ ("srow001", "Grace", "27"))
+
+ val query = source.toDF().toDF("key", "name", "age")
+ .writeStream
+ .format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ .option("catalog", streamCatalog)
+ .option(HBaseSparkConf.HBASE_CONFIG_LOCATION, configFile.getAbsolutePath)
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .trigger(Trigger.AvailableNow())
+ .start()
+
+ query.awaitTermination()
+
+ val result = loadStreamTable().orderBy("key").collect()
+ assert(result.length == 2)
+ assert(result(0).getAs[String]("key") == "srow000")
+ assert(result(0).getAs[String]("name") == "Frank")
+ assert(result(1).getAs[String]("key") == "srow001")
+ assert(result(1).getAs[String]("age") == "27")
+ }
+
+ test("streaming sink with short name 'hbase' alias") {
+ val ss = spark
+ import ss.implicits._
+
+ val checkpointDir =
Files.createTempDirectory("hbase-stream-alias-ckpt").toFile
+ checkpointDir.deleteOnExit()
+
+ implicit val sqlCtx = spark.sqlContext
+ val source = MemoryStream[(String, String, String)]
+
+ source.addData(("salias000", "Heidi", "41"))
+
+ val query = source.toDF().toDF("key", "name", "age")
+ .writeStream
+ .format("hbase")
+ .option("catalog", streamCatalog)
+ .option(HBaseSparkConf.HBASE_CONFIG_LOCATION, configFile.getAbsolutePath)
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
Review Comment:
This is addressed now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
