Add separate dag for recency tests (#500)

* Add dag for recency tests * export env variable * Update dbt image and schedule of recency dag * update variables * update dev image * update the schedule * lint * Update dags/dbt_recency_tests_dag.py Co-authored-by: chowbao <[email protected]> * update dbt image * update dag start date --------- Co-authored-by: chowbao <[email protected]>
stellar · Sep 25, 2024 · 5f7ad2a · 5f7ad2a
1 parent 464bc76
commit 5f7ad2a
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 2 deletions.
diff --git a/airflow_variables_dev.json b/airflow_variables_dev.json
@@ -123,7 +123,7 @@
     "partnership_assets__account_holders_activity_fact": false,
     "partnership_assets__asset_activity_fact": false
   },
-  "dbt_image_name": "stellar/stellar-dbt:02123b970",
+  "dbt_image_name": "stellar/stellar-dbt:90c8a57c1",
   "dbt_internal_source_db": "test-hubble-319619",
   "dbt_internal_source_schema": "test_crypto_stellar_internal",
   "dbt_job_execution_timeout_seconds": 300,
@@ -356,6 +356,7 @@
     "network_stats": 720,
     "ohlc": 720,
     "partnership_assets": 660,
+    "recency": 600,
     "relevant_asset_trades": 1200,
     "singular_test": 600,
     "snapshot_state": 600,

diff --git a/airflow_variables_prod.json b/airflow_variables_prod.json
@@ -124,7 +124,7 @@
     "partnership_assets__asset_activity_fact": false,
     "trade_agg": false
   },
-  "dbt_image_name": "stellar/stellar-dbt:02123b970",
+  "dbt_image_name": "stellar/stellar-dbt:90c8a57c1",
   "dbt_internal_source_db": "hubble-261722",
   "dbt_internal_source_schema": "crypto_stellar_internal_2",
   "dbt_job_execution_timeout_seconds": 2400,
@@ -354,6 +354,7 @@
     "network_stats": 360,
     "ohlc": 960,
     "partnership_assets": 1380,
+    "recency": 840,
     "relevant_asset_trades": 1800,
     "singular_test": 840,
     "snapshot_state": 840,

diff --git a/dags/dbt_recency_tests_dag.py b/dags/dbt_recency_tests_dag.py
@@ -0,0 +1,40 @@
+from datetime import datetime
+
+from airflow import DAG
+from airflow.operators.empty import EmptyOperator
+from kubernetes.client import models as k8s
+from stellar_etl_airflow.build_dbt_task import dbt_task
+from stellar_etl_airflow.build_elementary_slack_alert_task import elementary_task
+from stellar_etl_airflow.default import (
+    alert_sla_miss,
+    get_default_dag_args,
+    init_sentry,
+)
+
+init_sentry()
+
+with DAG(
+    "dbt_recency_tests",
+    default_args=get_default_dag_args(),
+    start_date=datetime(2024, 9, 24, 0, 0),
+    description="This DAG runs recency tests at a daily cadence",
+    schedule="0 20 * * *",  # Every day at 8:00 PM UTC / 3:00 PM CST
+    user_defined_filters={
+        "container_resources": lambda s: k8s.V1ResourceRequirements(requests=s),
+    },
+    max_active_runs=1,
+    catchup=False,
+    tags=["dbt-data-quality"],
+    # sla_miss_callback=alert_sla_miss,
+) as dag:
+
+    # DBT tests to run
+    recency_tests = dbt_task(
+        dag,
+        command_type="test",
+        tag="recency",
+        resource_cfg="dbt",
+        run_recency_test="true",
+    )
+
+    recency_tests
diff --git a/dags/stellar_etl_airflow/build_dbt_task.py b/dags/stellar_etl_airflow/build_dbt_task.py
@@ -68,6 +68,7 @@ def dbt_task(
     excluded=None,
     resource_cfg="default",
     run_singular_test="false",
+    run_recency_test="false",
 ):
     namespace = conf.get("kubernetes", "NAMESPACE")
     if namespace == "default":
@@ -136,6 +137,7 @@ def dbt_task(
             "EXECUTION_DATE": "{{ ts }}",
             "AIRFLOW_START_TIMESTAMP": "{{ ti.start_date.strftime('%Y-%m-%dT%H:%M:%SZ') }}",
             "IS_SINGULAR_AIRFLOW_TASK": run_singular_test,
+            "IS_RECENCY_AIRFLOW_TASK": run_recency_test,
         },
         image=dbt_image,
         arguments=args,