Configurable triggering interval from streaming ingestion (#63)

pyalex · web-flow · commit af0ec85bb017 · 2021-04-27T16:09:44.000+08:00
Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;

option default value

Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;
diff --git a/python/feast_spark/constants.py b/python/feast_spark/constants.py
@@ -160,6 +160,11 @@ class ConfigOptions(metaclass=ConfigMeta):
     #: Whitelisted Feast projects
     WHITELISTED_PROJECTS: Optional[str] = None
 
+    #: If set - streaming ingestion job will be consuming incoming rows not continuously,
+    #: but periodically with configured interval (in seconds).
+    #: That may help to control amount of write requests to storage
+    SPARK_STREAMING_TRIGGERING_INTERVAL: Optional[str] = None
+
     def defaults(self):
         return {
             k: getattr(self, k)
diff --git a/python/feast_spark/pyspark/abc.py b/python/feast_spark/pyspark/abc.py
@@ -503,6 +503,7 @@ def __init__(
         checkpoint_path: Optional[str] = None,
         stencil_url: Optional[str] = None,
         drop_invalid_rows: bool = False,
+        triggering_interval: Optional[int] = None,
     ):
         super().__init__(
             feature_table,
@@ -523,6 +524,7 @@ def __init__(
         )
         self._extra_jars = extra_jars
         self._checkpoint_path = checkpoint_path
+        self._triggering_interval = triggering_interval
 
     def get_name(self) -> str:
         return f"{self.get_job_type().to_pascal_case()}-{self.get_feature_table_name()}"
@@ -538,6 +540,8 @@ def get_arguments(self) -> List[str]:
         args.extend(["--mode", "online"])
         if self._checkpoint_path:
             args.extend(["--checkpoint-path", self._checkpoint_path])
+        if self._triggering_interval:
+            args.extend(["--triggering-interval", str(self._triggering_interval)])
         return args
 
     def get_job_hash(self) -> str:
diff --git a/python/feast_spark/pyspark/launcher.py b/python/feast_spark/pyspark/launcher.py
@@ -315,6 +315,9 @@ def get_stream_to_online_ingestion_params(
         checkpoint_path=client.config.get(opt.CHECKPOINT_PATH),
         stencil_url=client.config.get(opt.STENCIL_URL),
         drop_invalid_rows=client.config.get(opt.INGESTION_DROP_INVALID_ROWS),
+        triggering_interval=client.config.getint(
+            opt.SPARK_STREAMING_TRIGGERING_INTERVAL, default=None
+        ),
     )
 
 
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala b/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala
@@ -101,6 +101,9 @@ object IngestionJob {
 
     opt[String](name = "checkpoint-path")
       .action((x, c) => c.copy(checkpointPath = Some(x)))
+
+    opt[Int](name = "triggering-interval")
+      .action((x, c) => c.copy(streamingTriggeringSecs = x))
   }
 
   def main(args: Array[String]): Unit = {
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/bigtable/BigTableSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/bigtable/BigTableSinkRelation.scala
@@ -16,6 +16,8 @@
  */
 package feast.ingestion.stores.bigtable
 
+import java.io.IOException
+
 import com.google.cloud.bigtable.hbase.BigtableConfiguration
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hbase.{
@@ -75,7 +77,13 @@ class BigTableSinkRelation(
       try {
         admin.createTable(table)
       } catch {
-        case _: TableExistsException => admin.modifyTable(table)
+        case _: TableExistsException =>
+          try {
+            admin.modifyTable(table)
+          } catch {
+            case e: IOException =>
+              println(s"Table modification failed: ${e.getMessage}")
+          }
       }
     } finally {
       btConn.close()

Original file line number	Diff line number	Diff line change
`@@ -315,6 +315,9 @@ def get_stream_to_online_ingestion_params(`
`315`	`315`	`checkpoint_path=client.config.get(opt.CHECKPOINT_PATH),`
`316`	`316`	`stencil_url=client.config.get(opt.STENCIL_URL),`
`317`	`317`	`drop_invalid_rows=client.config.get(opt.INGESTION_DROP_INVALID_ROWS),`
	`318`	`+ triggering_interval=client.config.getint(`
	`319`	`+ opt.SPARK_STREAMING_TRIGGERING_INTERVAL, default=None`
	`320`	`+ ),`
`318`	`321`	`)`
`319`	`322`
`320`	`323`
Original file line number	Diff line number	Diff line change
`@@ -101,6 +101,9 @@ object IngestionJob {`
`101`	`101`
`102`	`102`	`opt[String](name = "checkpoint-path")`
`103`	`103`	`.action((x, c) => c.copy(checkpointPath = Some(x)))`
	`104`	`+`
	`105`	`+ opt[Int](name = "triggering-interval")`
	`106`	`+ .action((x, c) => c.copy(streamingTriggeringSecs = x))`
`104`	`107`	`}`
`105`	`108`
`106`	`109`	`def main(args: Array[String]): Unit = {`