import pendulum

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.providers.apache.hive.operators.hive import HiveOperator

with DAG(
    dag_id='hive_provider',
    schedule_interval=None,
    start_date=pendulum.datetime(2024, 10, 1, tz="UTC"),
    # 如果是True，配置定时运行的时候，它会补足所有缺少的任务
    catchup=False,
    tags=['demo'],
) as dag:
    start = DummyOperator(
        task_id="hive_start"
    )

    # 执行以下hql
    hive_operator = HiveOperator(
        task_id="hive_operator",
        hql="select * from test_db.person"
    )

    hive_operator2 = HiveOperator(
        task_id="hive_operator2",
        hql="select * from test_db.person limit 1"
    )

    hive_operator3 = HiveOperator(
        task_id="hive_operator3",
        hql="select * from test_db.person",
        hive_cli_conn_id="hiveserver2_default",
    )

    hive_post_action = BashOperator(
        task_id="hive_post_action",
        bash_command="echo 'last hql had been executed'",
        dag=dag
    )

    # start >> hive_operator >> hive_post_action
    # start >> hive_operator2 >> hive_post_action
    start >> [hive_operator, hive_operator2,hive_operator3] >> hive_post_action