add ScheduledBatchIngestionJobParameters

feast-dev · xiaoyongzhu · Apr 25, 2021 · Apr 26, 2021 · May 8, 2021 · May 19, 2021
commit 53d7e20e3176cb2202d9f7ceb50d90237166f9c5
diff --git a/python/feast_spark/pyspark/abc.py b/python/feast_spark/pyspark/abc.py
@@ -452,6 +452,62 @@ def get_arguments(self) -> List[str]:
             self._end.strftime("%Y-%m-%dT%H:%M:%S"),
         ]
 
+class ScheduledBatchIngestionJobParameters(IngestionJobParameters):
+    def __init__(
+        self,
+        feature_table: Dict,
+        source: Dict,
+        ingestion_timespan: int,
+        cron_schedule: str,
+        jar: str,
+        redis_host: Optional[str],
+        redis_port: Optional[int],
+        redis_ssl: Optional[bool],
+        bigtable_project: Optional[str],
+        bigtable_instance: Optional[str],
+        cassandra_host: Optional[str] = None,
+        cassandra_port: Optional[int] = None,
+        statsd_host: Optional[str] = None,
+        statsd_port: Optional[int] = None,
+        deadletter_path: Optional[str] = None,
+        stencil_url: Optional[str] = None,
+    ):
+        super().__init__(
+            feature_table,
+            source,
+            jar,
+            redis_host,
+            redis_port,
+            redis_ssl,
+            bigtable_project,
+            bigtable_instance,
+            cassandra_host,
+            cassandra_port,
+            statsd_host,
+            statsd_port,
+            deadletter_path,
+            stencil_url,
+        )
+        self._ingestion_timespan = ingestion_timespan
+        self._cron_schedule = cron_schedule
+
+    def get_name(self) -> str:
+        return f"{self.get_job_type().to_pascal_case()}-{self.get_feature_table_name()}"
+
+    def get_job_type(self) -> SparkJobType:
+        return SparkJobType.SCHEDULED_BATCH_INGESTION
+
+    def get_job_schedule(self) -> str:
+        return self._cron_schedule
+
+    def get_arguments(self) -> List[str]:
+        return super().get_arguments() + [
+            "--mode",
+            "offline",
+            "--ingestion-timespan",
+            str(self._ingestion_timespan),
+        ]
+
 
 class StreamIngestionJobParameters(IngestionJobParameters):
     def __init__(