[AI Flow Plugin] Solve job config parsing error bug (#624)

flink-extended · Sep 30, 2021 · 7cb73c1 · 7cb73c1
1 parent ac9261d
commit 7cb73c1
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 4 deletions.
diff --git a/flink-ai-flow/ai_flow_plugins/job_plugins/bash/bash_job_config.py b/flink-ai-flow/ai_flow_plugins/job_plugins/bash/bash_job_config.py
@@ -33,8 +33,9 @@ class BashJobConfig(JobConfig):
                     b: b
     """
     def __init__(self, job_name: Text = None,
+                 job_label_report_interval: float = 5.0,
                  properties: Dict[Text, Jsonable] = None) -> None:
-        super().__init__(job_name, 'bash', properties)
+        super().__init__(job_name, 'bash', job_label_report_interval, properties)
 
     @property
     def env(self):

diff --git a/flink-ai-flow/ai_flow_plugins/job_plugins/flink/flink_job_config.py b/flink-ai-flow/ai_flow_plugins/job_plugins/flink/flink_job_config.py
@@ -38,8 +38,9 @@ class FlinkJobConfig(JobConfig):
     """
 
     def __init__(self, job_name: Text = None,
+                 job_label_report_interval: float = 5.0,
                  properties: Dict[Text, Jsonable] = None) -> None:
-        super().__init__(job_name, 'flink', properties)
+        super().__init__(job_name, 'flink', job_label_report_interval, properties)
 
     @property
     def env(self):
@@ -81,4 +82,5 @@ def stop_mode(self):
     @classmethod
     def from_job_config(cls, job_config: JobConfig) -> 'FlinkJobConfig':
         return FlinkJobConfig(job_name=job_config.job_name,
+                              job_label_report_interval=job_config.job_label_report_interval,
                               properties=job_config.properties)
diff --git a/flink-ai-flow/ai_flow_plugins/job_plugins/python/python_job_config.py b/flink-ai-flow/ai_flow_plugins/job_plugins/python/python_job_config.py
@@ -34,8 +34,9 @@ class PythonJobConfig(JobConfig):
                     b: b
     """
     def __init__(self, job_name: Text = None,
+                 job_label_report_interval: float = 5.0,
                  properties: Dict[Text, Jsonable] = None) -> None:
-        super().__init__(job_name, 'python', properties)
+        super().__init__(job_name, 'python', job_label_report_interval, properties)
 
     @property
     def env(self):

diff --git a/flink-ai-flow/examples/wide_and_deep/workflows/wdl/cencus_stream_train_executors.py b/flink-ai-flow/examples/wide_and_deep/workflows/wdl/cencus_stream_train_executors.py
@@ -186,7 +186,7 @@ def process(self, execution_context: ExecutionContext, input_list: List[Table] =
         return []
 
 
-class StreamValidateExecutor(FlinkPythonProcessor):
+class StreamValidateExecutor(PythonProcessor):
     def __init__(self):
         super().__init__()
         self.path = None