(DRAFT) #1132 HERE dynamic binning aggregation

dags/here_dynamic_binning_agg.py

@@ -0,0 +1,87 @@
+'''
+To trigger for past date (range) use CLI:
+for i in {0..5}; do 
+    end_date=$(date -I -d "2023-11-02 +$i days")
+    airflow dags trigger -e "${end_date}" here_dynamic_binning_agg
+done
+
+or trigger just one day: airflow dags trigger -e 2023-11-02 here_dynamic_binning_agg
+`airflow dags backfill ...` doesn't work because there are no scheduled run dates in that range.
+'''
+
+import sys
+import os
+import logging
+from pendulum import duration, datetime
+
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.models import Variable
+from airflow.decorators import dag, task
+
+try:
+    repo_path = os.path.abspath(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
+    sys.path.insert(0, repo_path)
+    from dags.dag_functions import task_fail_slack_alert
+    from dags.custom_operators import SQLCheckOperatorWithReturnValue
+except:
+    raise ImportError("Cannot import slack alert functions")
+
+LOGGER = logging.getLogger(__name__)
+logging.basicConfig(level=logging.DEBUG)
+
+doc_md = "This DAG is running off the `1132-here-aggregation-proposal` branch to test dynamic binning aggregation."
+DAG_NAME = 'here_dynamic_binning_agg'
+DAG_OWNERS = Variable.get('dag_owners', deserialize_json=True).get(DAG_NAME, ["Unknown"]) 
+
+default_args = {
+    'owner': ','.join(DAG_OWNERS),
+    'depends_on_past':False,
+    'start_date': datetime(2019, 1, 1, tz="America/Toronto"),
+    'email_on_failure': False,
+    'email_on_success': False,
+    'retries': 1,
+    'retry_delay': duration(minutes=5),
+    #'on_failure_callback': task_fail_slack_alert
+}
+
+@dag(
+    DAG_NAME, 
+    default_args=default_args, 
+    schedule=None, # triggered by `pull_here_path` DAG
+    doc_md = doc_md,
+    tags=["HERE", "aggregation"],
+    max_active_runs=1,
+    catchup=False
+)
+
+#to add: catchup, one task at a time, depends on past.
+
+def here_dynamic_binning_agg():
+    check_not_empty = SQLCheckOperatorWithReturnValue(
+        task_id="check_not_empty",
+        sql="SELECT COUNT(*), COUNT(*) FROM here.ta_path WHERE dt = '{{ ds }}'",
+        conn_id="congestion_bot",
+        retries=1,
+        retry_delay=duration(days=1)
+    )
+
+    delete_daily = SQLExecuteQueryOperator(
+        sql="DELETE FROM gwolofs.congestion_raw_segments WHERE dt = '{{ ds }}'",
+        task_id='delete_daily',
+        conn_id='congestion_bot',
+        autocommit=True,
+        retries = 2
+    )
+
+    aggregate_daily = SQLExecuteQueryOperator(
+        sql="SELECT gwolofs.congestion_network_segment_agg('{{ ds }}'::date);",
+        task_id='aggregate_daily',
+        conn_id='congestion_bot',
+        autocommit=True,
+        retries = 2,
+        hook_params={"options": "-c statement_timeout=10800000ms"} #3 hours
+    )
+
+    check_not_empty >> delete_daily >> aggregate_daily
+
+here_dynamic_binning_agg()

dags/here_dynamic_binning_monthly_agg.py

@@ -0,0 +1,110 @@
+import os
+import sys
+import logging
+from pendulum import duration, datetime
+
+from airflow.models import Variable
+from airflow.decorators import dag, task
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.providers.postgres.hooks.postgres import PostgresHook
+
+try:
+    repo_path = os.path.abspath(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
+    sys.path.insert(0, repo_path)
+    from dags.dag_functions import task_fail_slack_alert
+    from dags.custom_operators import SQLCheckOperatorWithReturnValue
+except:
+    raise ImportError("Cannot import slack alert functions")
+
+LOGGER = logging.getLogger(__name__)
+logging.basicConfig(level=logging.DEBUG)
+
+doc_md = "This DAG is running off the `1132-here-aggregation-proposal` branch to test dynamic binning aggregation."
+DAG_NAME = 'here_dynamic_binning_monthly_agg'
+DAG_OWNERS = Variable.get('dag_owners', deserialize_json=True).get(DAG_NAME, ["Unknown"]) 
+
+default_args = {
+    'owner': ','.join(DAG_OWNERS),
+    'depends_on_past':False,
+    'start_date': datetime(2019, 1, 1, tz="America/Toronto"),
+    'retries': 1,
+    'retry_delay': duration(hours=1)
+    #'on_failure_callback': task_fail_slack_alert
+}
+
+@dag(
+    DAG_NAME,
+    default_args=default_args,
+    schedule='0 16 1 * *', # 4pm, first day of month
+    template_searchpath=os.path.join(repo_path,'here/traffic/sql/dynamic_bins'),
+    doc_md = doc_md,
+    tags=["HERE", "aggregation"],
+    max_active_runs=1,
+    catchup=True
+)
+
+#to add: catchup, one task at a time, depends on past.
+
+def here_dynamic_binning_monthly_agg():
+
+    check_missing_dates = SQLCheckOperatorWithReturnValue(
+        sql="select-check_missing_days.sql",
+        task_id="check_missing_dates",
+        conn_id='congestion_bot',
+        retries = 0
+    )
+
+    aggregate_monthly = SQLExecuteQueryOperator(
+        sql=[
+            "DELETE FROM gwolofs.congestion_segments_monthy_summary WHERE mnth = '{{ ds }}'",
+            "SELECT gwolofs.congestion_segment_monthly_agg('{{ ds }}')"
+        ],
+        task_id='aggregate_monthly',
+        conn_id='congestion_bot',
+        autocommit=True,
+        retries = 1
+    )
+
+    create_groups = SQLExecuteQueryOperator(
+        sql="segment_grouping.sql",
+        task_id="create_segment_groups",
+        conn_id='congestion_bot',
+        retries = 0,
+        params={"max_group_size": 100}
+    )
+
+    delete_data = SQLExecuteQueryOperator(
+        sql="DELETE FROM gwolofs.congestion_segments_monthly_bootstrap WHERE mnth = '{{ ds }}' AND n_resamples = 300",
+        task_id="delete_bootstrap_results",
+        conn_id='congestion_bot',
+        retries=0
+    )
+
+    @task
+    def expand_groups(**context):
+        return context["ti"].xcom_pull(task_ids="create_segment_groups")[0][0]
+
+    @task(retries=0, max_active_tis_per_dag=1)
+    def bootstrap_agg(segments, ds):
+        print(f"segments: {segments}")
+        postgres_cred = PostgresHook("congestion_bot")
+        query="""SELECT *
+            FROM UNNEST(%s::bigint[]) AS unnested(segment_id),
+            LATERAL (
+                SELECT gwolofs.congestion_segment_bootstrap(
+                                mnth := %s::date,
+                                segment_id := segment_id,
+                                n_resamples := 300)
+            ) AS lat"""
+        with postgres_cred.get_conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(query, (segments, ds))
+                conn.commit()
+
+    expand = expand_groups()
+
+    check_missing_dates >> aggregate_monthly >> create_groups >> delete_data
+    delete_data >> expand
+    bootstrap_agg.expand(segments=expand)
+
+here_dynamic_binning_monthly_agg()

dags/pull_here_path.py

@@ -3,10 +3,11 @@
 import pendulum
 from datetime import timedelta
 
-from airflow.decorators import task, dag
+from airflow.decorators import task, dag, task_group
 from airflow.hooks.base import BaseHook
 from airflow.models import Variable 
 from airflow.macros import ds_add, ds_format
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
 
 try:
     repo_path = os.path.abspath(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
@@ -45,7 +46,7 @@
 
 @dag(dag_id = dag_name,
      default_args=default_args,
-     schedule='30 10 * * *' ,
+     schedule='0 17 * * * ',
      catchup=False,
      doc_md = doc_md,
      tags=["HERE", "data_pull"]
@@ -85,6 +86,21 @@ def get_download_link(request_id: str, access_token: str):
     def load_data()->str:
         return '''curl $DOWNLOAD_URL | gunzip | psql -h $HOST -U $LOGIN -d bigdata -c "\\COPY here.ta_path_view FROM STDIN WITH (FORMAT csv, HEADER TRUE);" '''
 
-    load_data()
+    # Create a task group for triggering the DAGs
+    @task_group
+    def trigger_dags_tasks():
+        # Define TriggerDagRunOperator for each DAG to trigger
+        trigger_operators = []
+        DAGS_TO_TRIGGER = Variable.get('here_path_dag_triggers', deserialize_json=True)
+        for dag_id in DAGS_TO_TRIGGER:
+            trigger_operator = TriggerDagRunOperator(
+                task_id=f'trigger_{dag_id}',
+                trigger_dag_id=dag_id,
+                logical_date='{{macros.ds_add(ds, -1)}}',
+                reset_dag_run=True # Clear existing dag if already exists (for backfilling), old runs will not be in the logs
+            )
+            trigger_operators.append(trigger_operator)
+
+    load_data() >> trigger_dags_tasks()
 
 pull_here_path()

here/traffic/sql/dynamic_bins/corridor_agg.sql

@@ -0,0 +1,22 @@
+--test: 35 projects, 1 day = 47s
+SELECT
+    gwolofs.congestion_cache_tt_results_daily(
+        node_start := congestion_corridors.node_start,
+        node_end := congestion_corridors.node_end,
+        start_date := dates.dt::date
+    )
+FROM gwolofs.congestion_corridors
+JOIN gwolofs.congestion_projects USING (project_id),
+    generate_series('2025-01-01', '2025-02-28', '1 day'::interval) AS dates (dt)
+WHERE
+    congestion_projects.description IN (
+        'Avenue Road cycleway installation',
+        'bluetooth_corridors',
+        'scrutinized-cycleway-corridors'
+    )
+    AND corridor_id NOT IN (
+        SELECT DISTINCT corridor_id
+        FROM gwolofs.congestion_raw_corridors
+        WHERE dt >= '2025-01-01' AND dt < '2025-02-28'
+    )
+    AND map_version = '23_4';

here/traffic/sql/dynamic_bins/create-function-congestion_segment_bootstrap.sql

@@ -0,0 +1,91 @@
+--DROP FUNCTION gwolofs.congestion_segment_bootstrap(date,bigint,integer);
+
+CREATE OR REPLACE FUNCTION gwolofs.congestion_segment_bootstrap(
+    mnth date,
+    segment_id bigint,
+    n_resamples int
+)
+RETURNS void
+LANGUAGE SQL
+COST 100
+VOLATILE PARALLEL SAFE
+AS $BODY$
+
+    SELECT setseed(('0.'||replace(mnth::text, '-', ''))::numeric);
+
+    WITH raw_obs AS (
+        SELECT
+            --segment_id and mnth don't need to be in group by until end
+            EXTRACT('isodow' FROM dt) IN (1, 2, 3, 4, 5) AS is_wkdy,
+            hr,
+            ARRAY_AGG(tt::real) AS tt_array,
+            AVG(tt::real) AS avg_tt,
+            COUNT(*) AS n
+        FROM gwolofs.congestion_raw_segments
+        WHERE -- same params as the above aggregation
+            dt >= congestion_segment_bootstrap.mnth
+            AND dt < congestion_segment_bootstrap.mnth + interval '1 month'
+            AND segment_id = congestion_segment_bootstrap.segment_id
+        GROUP BY
+            segment_id,
+            is_wkdy,
+            hr
+    ),
+
+    random_selections AS (
+        SELECT
+            raw_obs.is_wkdy,
+            raw_obs.hr,
+            raw_obs.avg_tt,
+            raw_obs.n,
+            sample_group.group_id,
+            --get a random observation from the array of tts
+            AVG(raw_obs.tt_array[ceiling(random() * raw_obs.n)]) AS rnd_avg_tt
+        FROM raw_obs
+        CROSS JOIN generate_series(1, n)
+        -- 200 resamples (could be any number)
+        CROSS JOIN generate_series(1, congestion_segment_bootstrap.n_resamples) AS sample_group(group_id)
+        GROUP BY
+            raw_obs.is_wkdy,
+            raw_obs.hr,
+            raw_obs.avg_tt,
+            raw_obs.n,
+            sample_group.group_id
+    )
+
+    INSERT INTO gwolofs.congestion_segments_monthly_bootstrap (
+        segment_id, mnth, is_wkdy, hr, avg_tt, n, n_resamples, ci_lower, ci_upper
+    )
+    SELECT
+        congestion_segment_bootstrap.segment_id,
+        congestion_segment_bootstrap.mnth,
+        is_wkdy,
+        hr,
+        avg_tt::real,
+        n,
+        n_resamples,
+        percentile_disc(0.025) WITHIN GROUP (ORDER BY rnd_avg_tt)::real AS ci_lower,
+        percentile_disc(0.975) WITHIN GROUP (ORDER BY rnd_avg_tt)::real AS ci_upper
+    FROM random_selections
+    GROUP BY
+        is_wkdy,
+        hr,
+        avg_tt,
+        n;
+
+    $BODY$;
+
+GRANT EXECUTE ON FUNCTION gwolofs.congestion_segment_bootstrap(
+    date, bigint, integer
+) TO congestion_bot;
+
+/*Usage example: (works best one segment at a time with Lateral)
+SELECT *
+FROM UNNEST('{1,2,3,4,5,6,7,8,9}'::bigint[]) AS unnested(segment_id)
+LATERAL (
+    SELECT gwolofs.congestion_segment_bootstrap(
+                    mnth := '2025-06-01'::date,
+                    segment_ids := segment_id,
+                    n_resamples := 300)
+)
+*/