A join unifies related records from separate sources. In a financial personalization pipeline, it combines a financial snapshot with behavioral aggregates to create a customer feature that neither source reveals alone. section 13.1.2
Combining ledger and activity data
Pair daily P&L of −5% from the ledger with 10 logins from behavioral data to create a feature that may indicate panic. section 13.1.2
Listing 13.1 A conceptual Airflow DAG for the user mart pipeline
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_sql import SparkSqlOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from datetime import datetime
# 1. Define the DAG context
with DAG(
dag_id='user_behavior_mart_daily_etl',
start_date=datetime(2026, 1, 1),
schedule='0 2 * * *', # Run daily at 2:00 AM after market reconciliation
catchup=False,
doc_md="Generates the User Behavior Mart by joining logs and ledger."
) as dag:
# 2. Task: Aggregate Clickstream Logs (The "Noise")
# Uses Spark to crunch massive JSON logs from S3 into daily summaries.
aggregate_clickstream = SparkSqlOperator(
task_id='aggregate_clickstream_from_s3',
sql="""
INSERT OVERWRITE TABLE user_behavior_daily_aggregates
SELECT
user_id,
COUNT(CASE WHEN event = 'login' THEN 1 END) AS login_count_1d,
SUM(event_duration_sec) AS session_time_1d,
-- ... (omitted: complex window functions for 15d/30d) ...
FROM raw_app_logs
WHERE date = '{{ ds }}'
GROUP BY user_id;
"""
)
# 3. Task: Extract Financial Snapshot (The "Truth")
# Queries the transactional Postgres database for final EOD balances.
extract_financial_snapshot = SQLExecuteQueryOperator(
task_id='extract_financial_snapshot_from_postgres',
sql="""
INSERT INTO financial_daily_snapshots
SELECT
user_id, total_equity, cash_balance, daily_pnl_pct
FROM core_ledger_balances
WHERE snapshot_date = '{{ ds }}';
"""
)
# 4. Task: The "User 360" Join
# Merges the two streams to create the final feature store table.
build_final_mart = SparkSqlOperator(
task_id='build_final_user_mart',
sql="""
INSERT OVERWRITE TABLE user_behavior_mart
SELECT
f.user_id,
f.total_equity, f.daily_pnl_pct, -- Financial Context
b.login_count_1d, b.session_time_1d -- Behavioral Context
FROM financial_daily_snapshots f
JOIN user_behavior_daily_aggregates b
ON f.user_id = b.user_id
WHERE f.snapshot_date = '{{ ds }}';
"""
)
# 5. Define Dependencies
# Logs and Ledger are processed in parallel; the Join waits for both.
[aggregate_clickstream, extract_financial_snapshot] >> build_final_mart
See the conceptual Airflow DAG for a pipeline that aggregates clickstream logs and joins them with ledger data. section 13.1.2
