fix(SubPlat): Limit the bqetl_subplat_hourly DAG to one active run at a time (DENG-10213). (#8489)

Author: sean-rose

bigquery_etl/query_scheduling/dag.py

@@ -122,6 +122,7 @@ class Dag:
     repo: str = attr.ib("bigquery-etl")
     tags: List[str] = attr.ib([])
     catchup: bool = attr.ib(False)
+    max_active_runs: Optional[int] = attr.ib(None)
 
     @name.validator
     def validate_dag_name(self, attribute, value):

bigquery_etl/query_scheduling/templates/airflow_dag.j2

@@ -82,7 +82,19 @@ default_args = {{
 
 tags = {{ tags }}
 
-with DAG('{{ name }}', default_args=default_args{%+ if schedule_interval != None -%}, schedule_interval={{ schedule_interval | format_timedelta | format_schedule_interval }}{%+ endif -%}, doc_md=docs, tags=tags, catchup={{ catchup }}) as dag:
+with DAG(
+    '{{ name }}',
+    default_args=default_args,
+    {% if schedule_interval != None -%}
+    schedule_interval={{ schedule_interval | format_timedelta | format_schedule_interval }},
+    {%- endif %}
+    {% if max_active_runs -%}
+    max_active_runs={{ max_active_runs }},
+    {%- endif %}
+    doc_md=docs,
+    tags=tags,
+    catchup={{ catchup }},
+) as dag:
 {% for task_group in task_groups | sort %}
     task_group_{{ task_group }} = TaskGroup('{{ task_group }}')
 {% endfor %}

bigquery_etl/query_scheduling/templates/public_data_json_airflow_dag.j2

@@ -38,7 +38,18 @@ default_args = {{
 
 tags = {{ tags }}
 
-with DAG('{{ name }}', default_args=default_args{%+ if schedule_interval != None -%}, schedule_interval={{ schedule_interval | format_timedelta | format_schedule_interval }}{%+ endif -%}, doc_md = docs, tags = tags) as dag:
+with DAG(
+    '{{ name }}',
+    default_args=default_args,
+    {% if schedule_interval != None -%}
+    schedule_interval={{ schedule_interval | format_timedelta | format_schedule_interval }},
+    {%- endif %}
+    {% if max_active_runs -%}
+    max_active_runs={{ max_active_runs }},
+    {%- endif %}
+    doc_md=docs,
+    tags=tags,
+) as dag:
     docker_image = "gcr.io/moz-fx-data-airflow-prod-88e0/bigquery-etl:latest"
 {% for task_group in task_groups | sort %}
     task_group_{{ task_group }} = TaskGroup('{{ task_group }}')

dags.yaml

@@ -184,6 +184,8 @@ bqetl_subplat:
 
 bqetl_subplat_hourly:
   schedule_interval: 30 * * * *
+  # Limit concurrency to avoid any ETLs running while the Fivetran Stripe sync may be in progress.
+  max_active_runs: 1
   catchup: true
   description: |
     Hourly imports for Subscription Platform data from Stripe and Firestore,