Clean-up of google cloud example dags (#19436)

- Use start start_date - Use catchup=False - Tidy up the chaining of tasks in some cases
apache · Nov 7, 2021 · 9efb989 · 9efb989
1 parent 3a7e687
commit 9efb989
Show file tree

Hide file tree

Showing 24 changed files with 118 additions and 64 deletions.
diff --git a/airflow/providers/google/cloud/example_dags/example_automl_nl_text_classification.py b/airflow/providers/google/cloud/example_dags/example_automl_nl_text_classification.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -60,7 +60,8 @@
 with models.DAG(
  "example_automl_text_cls",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  tags=['example'],
 ) as example_dag:
  create_dataset_task = AutoMLCreateDatasetOperator(

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_nl_text_extraction.py b/airflow/providers/google/cloud/example_dags/example_automl_nl_text_extraction.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -59,7 +59,8 @@
 with models.DAG(
  "example_automl_text",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_nl_text_sentiment.py b/airflow/providers/google/cloud/example_dags/example_automl_nl_text_sentiment.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -60,7 +60,8 @@
 with models.DAG(
  "example_automl_text_sentiment",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_tables.py b/airflow/providers/google/cloud/example_dags/example_automl_tables.py
@@ -21,6 +21,7 @@
 """
 import os
 from copy import deepcopy
+from datetime import datetime
 from typing import Dict, List
 
 from airflow import models
@@ -40,7 +41,8 @@
  AutoMLTablesUpdateDatasetOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
+
+START_DATE = datetime(2021, 1, 1)
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -85,7 +87,8 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 with models.DAG(
  "example_create_and_deploy",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  user_defined_macros={
  "get_target_column_spec": get_target_column_spec,
  "target": TARGET,
@@ -197,7 +200,8 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 with models.DAG(
  "example_automl_dataset",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
 ) as example_dag:
  create_dataset_task = AutoMLCreateDatasetOperator(
@@ -265,7 +269,8 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 with models.DAG(
  "example_gcp_get_deploy",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  tags=["example"],
 ) as get_deploy_dag:
  # [START howto_operator_get_model]
@@ -290,7 +295,8 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 with models.DAG(
  "example_gcp_predict",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  tags=["example"],
 ) as predict_dag:
  # [START howto_operator_prediction]

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_translation.py b/airflow/providers/google/cloud/example_dags/example_automl_translation.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -66,7 +66,8 @@
 with models.DAG(
  "example_automl_translation",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/...w/providers/google/cloud/example_dags/example_automl_video_intelligence_classification.py b/...w/providers/google/cloud/example_dags/example_automl_video_intelligence_classification.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -63,7 +63,8 @@
 with models.DAG(
  "example_automl_video",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_video_intelligence_tracking.py b/airflow/providers/google/cloud/example_dags/example_automl_video_intelligence_tracking.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -64,7 +64,8 @@
 with models.DAG(
  "example_automl_video_tracking",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_vision_classification.py b/airflow/providers/google/cloud/example_dags/example_automl_vision_classification.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -61,7 +61,8 @@
 with models.DAG(
  "example_automl_vision",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_automl_vision_object_detection.py b/airflow/providers/google/cloud/example_dags/example_automl_vision_object_detection.py
@@ -20,6 +20,7 @@
 Example Airflow DAG that uses Google AutoML services.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.hooks.automl import CloudAutoMLHook
@@ -30,7 +31,6 @@
  AutoMLImportDataOperator,
  AutoMLTrainModelOperator,
 )
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-project-id")
 GCP_AUTOML_LOCATION = os.environ.get("GCP_AUTOML_LOCATION", "us-central1")
@@ -63,7 +63,8 @@
 with models.DAG(
  "example_automl_vision_detection",
  schedule_interval=None, # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  user_defined_macros={"extract_object_id": extract_object_id},
  tags=['example'],
 ) as example_dag:

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_dts.py b/airflow/providers/google/cloud/example_dags/example_bigquery_dts.py
@@ -21,6 +21,7 @@
 """
 import os
 import time
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.operators.bigquery_dts import (
@@ -29,7 +30,6 @@
  BigQueryDeleteDataTransferConfigOperator,
 )
 from airflow.providers.google.cloud.sensors.bigquery_dts import BigQueryDataTransferServiceTransferRunSensor
-from airflow.utils.dates import days_ago
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 BUCKET_URI = os.environ.get("GCP_DTS_BUCKET_URI", "gs://INVALID BUCKET NAME/bank-marketing.csv")
@@ -65,7 +65,8 @@
 with models.DAG(
  "example_gcp_bigquery_dts",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  tags=['example'],
 ) as dag:
  # [START howto_bigquery_create_data_transfer]

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_operations.py b/airflow/providers/google/cloud/example_dags/example_bigquery_operations.py
@@ -21,6 +21,7 @@
 """
 import os
 import time
+from datetime import datetime
 from urllib.parse import urlparse
 
 from airflow import models
@@ -38,7 +39,8 @@
  BigQueryUpdateTableSchemaOperator,
  BigQueryUpsertTableOperator,
 )
-from airflow.utils.dates import days_ago
+
+START_DATE = datetime(2021, 1, 1)
 
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 BQ_LOCATION = "europe-north1"
@@ -58,7 +60,8 @@
 with models.DAG(
  "example_bigquery_operations",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  tags=["example"],
 ) as dag:
  # [START howto_operator_bigquery_create_table]
@@ -238,7 +241,8 @@
 with models.DAG(
  "example_bigquery_operations_location",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=START_DATE,
+ catchup=False,
  tags=["example"],
 ) as dag_with_location:
  create_dataset_with_location = BigQueryCreateEmptyDatasetOperator(

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_queries.py b/airflow/providers/google/cloud/example_dags/example_bigquery_queries.py
@@ -34,7 +34,6 @@
  BigQueryIntervalCheckOperator,
  BigQueryValueCheckOperator,
 )
-from airflow.utils.dates import days_ago
 
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 DATASET_NAME = os.environ.get("GCP_BIGQUERY_DATASET_NAME", "test_dataset")
@@ -65,7 +64,8 @@
  with models.DAG(
  dag_id,
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  tags=["example"],
  user_defined_macros={"DATASET": DATASET, "TABLE": TABLE_1},
  ) as dag_with_locations:

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_sensors.py b/airflow/providers/google/cloud/example_dags/example_bigquery_sensors.py
@@ -33,7 +33,6 @@
  BigQueryTableExistenceSensor,
  BigQueryTablePartitionExistenceSensor,
 )
-from airflow.utils.dates import days_ago
 
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 DATASET_NAME = os.environ.get("GCP_BIGQUERY_DATASET_NAME", "test_sensors_dataset")
@@ -55,7 +54,8 @@
 with models.DAG(
  dag_id,
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  tags=["example"],
  user_defined_macros={"DATASET": DATASET_NAME, "TABLE": TABLE_NAME},
  default_args={"project_id": PROJECT_ID},

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_to_bigquery.py b/airflow/providers/google/cloud/example_dags/example_bigquery_to_bigquery.py
@@ -20,6 +20,7 @@
 Example Airflow DAG for Google BigQuery service.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.operators.bigquery import (
@@ -28,7 +29,6 @@
  BigQueryDeleteDatasetOperator,
 )
 from airflow.providers.google.cloud.transfers.bigquery_to_bigquery import BigQueryToBigQueryOperator
-from airflow.utils.dates import days_ago
 
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 DATASET_NAME = os.environ.get("GCP_BIGQUERY_DATASET_NAME", "test_dataset_transfer")
@@ -38,7 +38,8 @@
 with models.DAG(
  "example_bigquery_to_bigquery",
  schedule_interval='@once', # Override to match your needs
- start_date=days_ago(1),
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
  tags=["example"],
 ) as dag:
  copy_selected_data = BigQueryToBigQueryOperator(