Deprecate airflow_runner.AirflowDAGRunner and runner.KubeflowRunner after 0.14 release

PiperOrigin-RevId: 266229885

Author: tfx-copybara

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_kubeflow_test.py

@@ -20,7 +20,7 @@
 import os
 import tensorflow as tf
 from tfx.examples.chicago_taxi_pipeline import taxi_pipeline_kubeflow
-from tfx.orchestration.kubeflow.runner import KubeflowRunner
+from tfx.orchestration.kubeflow.kubeflow_dag_runner import KubeflowDagRunner
 
 
 class TaxiPipelineKubeflowTest(tf.test.TestCase):
@@ -50,7 +50,7 @@ def testTaxiPipelineConstructionAndDefinitionFileExists(self):
         ._ai_platform_serving_args)
     self.assertEqual(9, len(logical_pipeline.components))
 
-    KubeflowRunner().run(logical_pipeline)
+    KubeflowDagRunner().run(logical_pipeline)
     file_path = os.path.join(self._tmp_dir,
                              'chicago_taxi_pipeline_kubeflow.tar.gz')
     self.assertTrue(tf.gfile.Exists(file_path))

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_mysql.py

@@ -32,7 +32,7 @@
 from tfx.components.transform.component import Transform
 from tfx.orchestration import metadata
 from tfx.orchestration import pipeline
-from tfx.orchestration.airflow.airflow_runner import AirflowDAGRunner
+from tfx.orchestration.airflow.airflow_dag_runner import AirflowDagRunner
 from tfx.proto import evaluator_pb2
 from tfx.proto import pusher_pb2
 from tfx.proto import trainer_pb2
@@ -143,7 +143,8 @@ def _create_pipeline(
       metadata_connection_config=metadata_connection_config)
 
 
-airflow_pipeline = AirflowDAGRunner(_airflow_config).run(
+# 'DAG' below need to be kept for Airflow to detect dag.
+DAG = AirflowDagRunner(_airflow_config).run(
     _create_pipeline(
         pipeline_name=_pipeline_name,
         pipeline_root=_pipeline_root,

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_portable_beam.py

@@ -31,7 +31,7 @@
 from tfx.components.transform.component import Transform
 from tfx.orchestration import metadata
 from tfx.orchestration import pipeline
-from tfx.orchestration.airflow.airflow_runner import AirflowDAGRunner
+from tfx.orchestration.airflow.airflow_dag_runner import AirflowDagRunner
 from tfx.proto import evaluator_pb2
 from tfx.proto import pusher_pb2
 from tfx.proto import trainer_pb2
@@ -159,12 +159,14 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,
               # TODO(FLINK-10672): Obviate setting BATCH_FORCED.
               '--execution_mode_for_batch=BATCH_FORCED',
           ],
-          # LINT.ThenChange(tfx/examples/chicago_taxi/setup_beam_on_portable_beam.sh)
+          # LINT.ThenChange(../chicago_taxi/setup_beam_on_spark.sh)
+          # LINT.ThenChange(../chicago_taxi/setup_beam_on_flink.sh)
       })
 
 
 # TODO(jyzhao): consider using beam orchestrator after b/137294896.
-airflow_pipeline = AirflowDAGRunner(_airflow_config).run(
+# 'DAG' below need to be kept for Airflow to detect dag.
+DAG = AirflowDagRunner(_airflow_config).run(
     _create_pipeline(
         pipeline_name=_pipeline_name,
         pipeline_root=_pipeline_root,

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple.py

@@ -31,7 +31,7 @@
 from tfx.components.transform.component import Transform
 from tfx.orchestration import metadata
 from tfx.orchestration import pipeline
-from tfx.orchestration.airflow.airflow_runner import AirflowDAGRunner
+from tfx.orchestration.airflow.airflow_dag_runner import AirflowDagRunner
 from tfx.proto import evaluator_pb2
 from tfx.proto import pusher_pb2
 from tfx.proto import trainer_pb2
@@ -137,7 +137,8 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,
           metadata_path))
 
 
-airflow_pipeline = AirflowDAGRunner(_airflow_config).run(
+# 'DAG' below need to be kept for Airflow to detect dag.
+DAG = AirflowDagRunner(_airflow_config).run(
     _create_pipeline(
         pipeline_name=_pipeline_name,
         pipeline_root=_pipeline_root,

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple_test.py

@@ -24,7 +24,7 @@
 import tensorflow as tf
 
 from tfx.examples.chicago_taxi_pipeline import taxi_pipeline_simple
-from tfx.orchestration.airflow.airflow_runner import AirflowDAGRunner
+from tfx.orchestration.airflow.airflow_dag_runner import AirflowDagRunner
 
 
 class TaxiPipelineSimpleTest(tf.test.TestCase):
@@ -48,7 +48,7 @@ def testTaxiPipelineCheckDagConstruction(self):
         serving_model_dir=self._test_dir,
         metadata_path=self._test_dir)
     self.assertEqual(9, len(logical_pipeline.components))
-    pipeline = AirflowDAGRunner(airflow_config).run(logical_pipeline)
+    pipeline = AirflowDagRunner(airflow_config).run(logical_pipeline)
     self.assertIsInstance(pipeline, models.DAG)
 
 

tfx/orchestration/airflow/airflow_runner.py

@@ -17,7 +17,10 @@
 from __future__ import division
 from __future__ import print_function
 
+from tensorflow.python.util import deprecation  # pylint: disable=g-direct-tensorflow-import
 from tfx.orchestration.airflow import airflow_dag_runner
 
-# TODO(jyzhao): deprecate this and change taxi examples.
-AirflowDAGRunner = airflow_dag_runner.AirflowDagRunner
+AirflowDAGRunner = deprecation.deprecated_alias(  # pylint: disable=invalid-name
+    deprecated_name='airflow_runner.AirflowDAGRunner',
+    name='airflow_dag_runner.AirflowDagRunner',
+    func_or_class=airflow_dag_runner.AirflowDagRunner)

tfx/orchestration/kubeflow/runner.py

@@ -17,8 +17,14 @@
 from __future__ import division
 from __future__ import print_function
 
+from tensorflow.python.util import deprecation  # pylint: disable=g-direct-tensorflow-import
 from tfx.orchestration.kubeflow import kubeflow_dag_runner
 
-# TODO(jyzhao): deprecate this and change taxi examples.
-KubeflowRunner = kubeflow_dag_runner.KubeflowDagRunner
-KubeflowRunnerConfig = kubeflow_dag_runner.KubeflowDagRunnerConfig
+KubeflowRunner = deprecation.deprecated_alias(  # pylint: disable=invalid-name
+    deprecated_name='runner.KubeflowRunner',
+    name='kubeflow_dag_runner.KubeflowDagRunner',
+    func_or_class=kubeflow_dag_runner.KubeflowDagRunner)
+KubeflowRunnerConfig = deprecation.deprecated_alias(  # pylint: disable=invalid-name
+    deprecated_name='runner.KubeflowRunnerConfig',
+    name='kubeflow_dag_runner.KubeflowDagRunnerConfig',
+    func_or_class=kubeflow_dag_runner.KubeflowDagRunnerConfig)