Generate avro files from bq tables

.sqlfluff

@@ -81,6 +81,7 @@ batch_id = batch_id
 batch_run_date = batch_run_date
 prev_batch_run_date = prev_batch_run_date
 next_batch_run_date = next_batch_run_date
+uri = uri
 
 # Some rules can be configured directly from the config common to other rules
 [sqlfluff:rules]

airflow_variables_dev.json

@@ -1,5 +1,6 @@
 {
   "api_key_path": "/home/airflow/gcs/data/apiKey.json",
+  "avro_gcs_bucket": "test_dune_bucket_sdf",
   "bq_dataset": "test_crypto_stellar_internal",
   "bq_dataset_audit_log": "audit_log",
   "bq_project": "test-hubble-319619",
@@ -333,6 +334,7 @@
   "task_sla": {
     "asset_stats": 720,
     "build_batch_stats": 840,
+    "build_bq_generate_avro_job": 600,
     "build_bq_insert_job": 1080,
     "build_del_ins_from_gcs_to_bq_task": 2000,
     "build_delete_data_task": 1020,
@@ -366,6 +368,7 @@
   },
   "task_timeout": {
     "build_batch_stats": 180,
+    "build_bq_generate_avro_job": 600,
     "build_bq_insert_job": 180,
     "build_copy_table": 180,
     "build_dbt_task": 960,

airflow_variables_prod.json

@@ -1,5 +1,6 @@
 {
   "api_key_path": "/home/airflow/gcs/data/apiKey.json",
+  "avro_gcs_bucket": "dune_bucket_sdf",
   "bq_dataset": "crypto_stellar_internal_2",
   "bq_dataset_audit_log": "audit_log",
   "bq_project": "hubble-261722",
@@ -331,6 +332,7 @@
   "task_sla": {
     "asset_stats": 420,
     "build_batch_stats": 600,
+    "build_bq_generate_avro_job": 600,
     "build_bq_insert_job": 840,
     "build_del_ins_from_gcs_to_bq_task": 2000,
     "build_delete_data_task": 780,
@@ -364,6 +366,7 @@
   },
   "task_timeout": {
     "build_batch_stats": 180,
+    "build_bq_generate_avro_job": 600,
     "build_bq_insert_job": 180,
     "build_copy_table": 180,
     "build_dbt_task": 1800,

dags/generate_avro_files_dag.py

@@ -0,0 +1,75 @@
+from datetime import datetime
+
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+from stellar_etl_airflow import macros
+from stellar_etl_airflow.build_bq_generate_avro_job_task import (
+    build_bq_generate_avro_job,
+)
+from stellar_etl_airflow.build_cross_dependency_task import build_cross_deps
+from stellar_etl_airflow.default import (
+    alert_sla_miss,
+    get_default_dag_args,
+    init_sentry,
+)
+
+init_sentry()
+
+dag = DAG(
+    "generate_avro",
+    default_args=get_default_dag_args(),
+    start_date=datetime(2024, 10, 1, 1, 0),
+    catchup=True,
+    description="This DAG generates AVRO files from BQ tables",
+    schedule_interval="0 * * * *",
+    render_template_as_native_obj=True,
+    user_defined_macros={
+        "subtract_data_interval": macros.subtract_data_interval,
+        "batch_run_date_as_datetime_string": macros.batch_run_date_as_datetime_string,
+        "batch_run_date_as_directory_string": macros.batch_run_date_as_directory_string,
+    },
+    sla_miss_callback=alert_sla_miss,
+)
+
+public_project = "{{ var.value.public_project }}"
+public_dataset = "{{ var.value.public_dataset }}"
+gcs_bucket = "{{ var.value.avro_gcs_bucket }}"
+
+
+# Wait on ingestion DAGs
+wait_on_history_table = build_cross_deps(
+    dag, "wait_on_ledgers_txs", "history_table_export"
+)
+wait_on_state_table = build_cross_deps(dag, "wait_on_state_table", "state_table_export")
+
+# Add dummy_task so DAG generates the avro_tables loop and dependency graph correctly
+dummy_task = DummyOperator(task_id="dummy_task", dag=dag)
+
+# Generate AVRO files
+avro_tables = [
+    "accounts",
+    "contract_data",
+    "history_contract_events",
+    "history_ledgers",
+    "history_trades",
+    "history_transactions",
+    "liquidity_pools",
+    "offers",
+    "trust_lines",
+    "ttl",
+    # "history_effects",
+    # "history_operations",
+]
+
+for table in avro_tables:
+    avro_task = build_bq_generate_avro_job(
+        dag=dag,
+        project=public_project,
+        dataset=public_dataset,
+        table=table,
+        gcs_bucket=gcs_bucket,
+    )
+
+    dummy_task >> avro_task
+    wait_on_history_table >> avro_task
+    wait_on_state_table >> avro_task

dags/queries/generate_avro/accounts.sql

@@ -0,0 +1,20 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (sequence_ledger, batch_id, batch_insert_ts, batch_run_date)
+        , sequence_ledger as account_sequence_last_modified_ledger
+    from {project_id}.{dataset_id}.accounts
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/contract_data.sql

@@ -0,0 +1,17 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.contract_data
+    where
+        true
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_contract_events.sql

@@ -0,0 +1,17 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.history_contract_events
+    where
+        true
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_effects.sql

@@ -0,0 +1,21 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (details, batch_id, batch_insert_ts, batch_run_date)
+        , details.*
+        except (predicate)
+    from {project_id}.{dataset_id}.history_effects
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_ledgers.sql

@@ -0,0 +1,17 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.history_ledgers
+    where
+        true
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_operations.sql

@@ -0,0 +1,22 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (details, details_json, batch_id, batch_insert_ts, batch_run_date)
+        , details.*
+        except (claimants, type)
+        , details.type as soroban_operation_type
+    from {project_id}.{dataset_id}.history_operations
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_trades.sql

@@ -0,0 +1,18 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (ledger_closed_at, batch_id, batch_insert_ts, batch_run_date)
+        , ledger_closed_at as closed_at
+    from {project_id}.{dataset_id}.history_trades
+    where
+        true
+        and ledger_closed_at >= '{batch_run_date}'
+        and ledger_closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/history_transactions.sql

@@ -0,0 +1,19 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.history_transactions
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/liquidity_pools.sql

@@ -0,0 +1,19 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.liquidity_pools
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/offers.sql

@@ -0,0 +1,19 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.offers
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/trust_lines.sql

@@ -0,0 +1,19 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.trust_lines
+    where
+        true
+        and batch_run_date >= '{batch_run_date}'
+        and batch_run_date < '{next_batch_run_date}'
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)

dags/queries/generate_avro/ttl.sql

@@ -0,0 +1,17 @@
+export data
+options (
+    uri = '{uri}'
+    , format = 'avro'
+    , overwrite = true
+)
+as (
+    select
+        *
+        except (batch_id, batch_insert_ts, batch_run_date)
+    from {project_id}.{dataset_id}.ttl
+    where
+        true
+        and closed_at >= '{batch_run_date}'
+        and closed_at < '{next_batch_run_date}'
+    order by closed_at asc
+)