Fixup docstring for deprecated DataprocSubmitSparkJobOperator and refactoring system tests (#32743)

Author: moiseenkov

airflow/providers/google/cloud/operators/dataproc.py

@@ -1384,6 +1384,10 @@ def execute(self, context: Context):
 class DataprocSubmitSparkJobOperator(DataprocJobBaseOperator):
     """Start a Spark Job on a Cloud DataProc cluster.
 
+    .. seealso::
+        This operator is deprecated, please use
+        :class:`~airflow.providers.google.cloud.operators.dataproc.DataprocSubmitJobOperator`:
+
     :param main_jar: The HCFS URI of the jar file that contains the main class
         (use this or the main_class, not both together).
     :param main_class: Name of the job class. (use this or the main_jar, not both

tests/system/providers/google/cloud/dataproc/example_dataproc_spark.py

@@ -35,10 +35,8 @@
 DAG_ID = "dataproc_spark"
 PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT")
 
-CLUSTER_NAME = f"cluster-dataproc-spark-{ENV_ID}"
+CLUSTER_NAME = f"cluster-{ENV_ID}-{DAG_ID}".replace("_", "-")
 REGION = "europe-west1"
-ZONE = "europe-west1-b"
-
 
 # Cluster definition
 CLUSTER_CONFIG = {
@@ -54,8 +52,6 @@
     },
 }
 
-TIMEOUT = {"seconds": 1 * 24 * 60 * 60}
-
 # Jobs definitions
 # [START how_to_cloud_dataproc_spark_config]
 SPARK_JOB = {
@@ -74,7 +70,7 @@
     schedule="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
-    tags=["example", "dataproc"],
+    tags=["example", "dataproc", "spark"],
 ) as dag:
     create_cluster = DataprocCreateClusterOperator(
         task_id="create_cluster",
@@ -96,7 +92,14 @@
         trigger_rule=TriggerRule.ALL_DONE,
     )
 
-    create_cluster >> spark_task >> delete_cluster
+    (
+        # TEST SETUP
+        create_cluster
+        # TEST BODY
+        >> spark_task
+        # TEST TEARDOWN
+        >> delete_cluster
+    )
 
     from tests.system.utils.watcher import watcher
 

tests/system/providers/google/cloud/dataproc/example_dataproc_spark_async.py

@@ -36,9 +36,8 @@
 DAG_ID = "dataproc_spark_async"
 PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT")
 
-CLUSTER_NAME = f"dataproc-spark-async-{ENV_ID}"
+CLUSTER_NAME = f"cluster-{ENV_ID}-{DAG_ID}".replace("_", "-")
 REGION = "europe-west1"
-ZONE = "europe-west1-b"
 
 # Cluster definition
 CLUSTER_CONFIG = {
@@ -54,8 +53,6 @@
     },
 }
 
-TIMEOUT = {"seconds": 1 * 24 * 60 * 60}
-
 # Jobs definitions
 SPARK_JOB = {
     "reference": {"project_id": PROJECT_ID},
@@ -72,7 +69,7 @@
     schedule="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
-    tags=["example", "dataproc"],
+    tags=["example", "dataproc", "spark", "async"],
 ) as dag:
     create_cluster = DataprocCreateClusterOperator(
         task_id="create_cluster",
@@ -104,7 +101,15 @@
         trigger_rule=TriggerRule.ALL_DONE,
     )
 
-    create_cluster >> spark_task_async >> spark_task_async_sensor >> delete_cluster
+    (
+        # TEST SETUP
+        create_cluster
+        # TEST BODY
+        >> spark_task_async
+        >> spark_task_async_sensor
+        # TEST TEARDOWN
+        >> delete_cluster
+    )
 
     from tests.system.utils.watcher import watcher
 

tests/system/providers/google/cloud/dataproc/example_dataproc_spark_deferrable.py

@@ -36,10 +36,8 @@
 DAG_ID = "dataproc_spark_deferrable"
 PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT")
 
-CLUSTER_NAME = f"cluster-dataproc-spark-{ENV_ID}"
+CLUSTER_NAME = f"cluster-{ENV_ID}-{DAG_ID}".replace("_", "-")
 REGION = "europe-west1"
-ZONE = "europe-west1-b"
-
 
 # Cluster definition
 CLUSTER_CONFIG = {
@@ -55,8 +53,6 @@
     },
 }
 
-TIMEOUT = {"seconds": 1 * 24 * 60 * 60}
-
 # Jobs definitions
 # [START how_to_cloud_dataproc_spark_deferrable_config]
 SPARK_JOB = {
@@ -75,7 +71,7 @@
     schedule_interval="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
-    tags=["example", "dataproc"],
+    tags=["example", "dataproc", "spark", "deferrable"],
 ) as dag:
     create_cluster = DataprocCreateClusterOperator(
         task_id="create_cluster",
@@ -97,7 +93,14 @@
         trigger_rule=TriggerRule.ALL_DONE,
     )
 
-    create_cluster >> spark_task >> delete_cluster
+    (
+        # TEST SETUP
+        create_cluster
+        # TEST BODY
+        >> spark_task
+        # TEST TEARDOWN
+        >> delete_cluster
+    )
 
     from tests.system.utils.watcher import watcher