Copilot commented on code in PR #168:
URL: https://github.com/apache/hbase-connectors/pull/168#discussion_r4086988351
##########
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 test does not set up `test_stream_sink` itself; it only succeeds when
the preceding test has already created that table. Running this test alone (or
with a different test order) reaches the existing-table write path with a
missing HBase table and fails. Create/use a distinct table for this test or
pass the table-creation option so it is independently runnable.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseStreamingDataWriterFactory.scala:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.DataWriter
+import
org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory
+import org.apache.spark.sql.types.StructType
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * This is a new class in the spark4 module. Implements
StreamingDataWriterFactory for the DS V2
+ * streaming write path. Serialized to executors. Creates one HBaseDataWriter
per Spark partition
+ * per micro-batch epoch. The epochId is not used because HBase puts are
idempotent — a retried
+ * epoch writes the same rows with the same row keys, producing no duplicates.
+ *
+ * @param schema
+ * @param properties
+ * @param catalog
+ * @param wrappedConf
+ */
[email protected]
+class HBaseStreamingDataWriterFactory(
+ schema: StructType,
+ properties: Map[String, String],
+ catalog: HBaseTableCatalog,
+ wrappedConf: SerializableConfiguration)
+ extends StreamingDataWriterFactory
+ with Serializable {
+
+ override def createWriter(
+ partitionId: Int,
+ taskId: Long,
+ epochId: Long): DataWriter[InternalRow] = {
+ new HBaseDataWriter(schema, properties, catalog, wrappedConf)
Review Comment:
The factory discards `epochId`, but `HBaseDataWriter` creates each default
`Put` with a server-assigned timestamp. If Spark retries an epoch, the same
row/cells are written again as new HBase versions, so the idempotence claim in
the factory's documentation is not true for versioned reads and can increase
version/storage churn. Pass epoch identity through and use a deliberate
deduplication or stable-timestamp strategy, or document the sink as
at-least-once rather than idempotent.
--
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]
