airflow_dag_druid_ingestion_and_data_check_operators.py

1import pandas as pd
2from datetime import datetime
3from airflow.models import DAG
4from airflow.providers.apache.druid.operators.druid import DruidOperator
5from airflow.providers.apache.druid.operators.druid_check import DruidCheckOperator
6
7with DAG(
8    dag_id='example_druid_operator',
9    schedule=None,
10    start_date=datetime(2021, 1, 1),
11    catchup=False,
12    tags=['example'],
13) as dag:
14
15    # Example of submitting a JSON ingestion spec to Druid
16    submit_job = DruidOperator(
17        task_id='submit_druid_ingestion',
18        json_index_file='path/to/ingestion_spec.json',
19        druid_ingest_conn_id='druid_ingest_default'
20    )
21
22    # Example of checking if a specific metric meets a threshold in Druid
23    check_query = DruidCheckOperator(
24        task_id='check_druid_data',
25        sql='SELECT COUNT(*) FROM my_datasource WHERE __time >= CURRENT_TIMESTAMP - INTERVAL "1" DAY',
26        druid_broker_conn_id='druid_broker_default'
27    )
28
29    submit_job >> check_query