你不可不知的任务调度神器 -AirFlow

# airflow 需要 home 目录，默认是~/airflow，# 但是如果你需要，放在其它位置也是可以的# (可选)export AIRFLOW_HOME = ~/airflow
# 使用 pip 从 pypi 安装pip install apache-airflow
# 初始化数据库airflow initdb
# 启动 web 服务器，默认端口是 8080airflow webserver -p 8080
# 启动定时器airflow scheduler
# 在浏览器中浏览 localhost:8080，并在 home 页开启 example dag

# The executor class that airflow should use. Choices include# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor, KubernetesExecutorexecutor = LocalExecutor# The SqlAlchemy connection string to the metadata database.# SqlAlchemy supports many different database engine, more information# their websitesql_alchemy_conn = mysql://root:xxxxxx@localhost:3306/airflow

"""### ETL DAG Tutorial DocumentationThis ETL DAG is compatible with Airflow 1.10.x (specifically tested with 1.10.12) and is referencedas part of the documentation that goes along with the Airflow Functional DAG tutorial located[here](https://airflow.apache.org/tutorial_decorated_flows.html)"""# [START tutorial]# [START import_module]import json
# The DAG object; we'll need this to instantiate a DAGfrom airflow import DAG
# Operators; we need this to operate!from airflow.operators.python import PythonOperatorfrom airflow.utils.dates import days_ago
# [END import_module]
# [START default_args]# These args will get passed on to each operator# You can override them on a per-task basis during operator initializationdefault_args = {    'owner': 'airflow',}# [END default_args]
# [START instantiate_dag]with DAG(    'tutorial_etl_dag',    default_args=default_args,    description='ETL DAG tutorial',    schedule_interval=None,    start_date=days_ago(2),    tags=['example'],) as dag:    # [END instantiate_dag]    # [START documentation]    dag.doc_md = __doc__    # [END documentation]
    # [START extract_function]    def extract(**kwargs):        ti = kwargs['ti']        data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'        ti.xcom_push('order_data', data_string)
    # [END extract_function]
    # [START transform_function]    def transform(**kwargs):        ti = kwargs['ti']        extract_data_string = ti.xcom_pull(task_ids='extract', key='order_data')        order_data = json.loads(extract_data_string)
        total_order_value = 0        for value in order_data.values():            total_order_value += value
        total_value = {"total_order_value": total_order_value}        total_value_json_string = json.dumps(total_value)        ti.xcom_push('total_order_value', total_value_json_string)
    # [END transform_function]
    # [START load_function]    def load(**kwargs):        ti = kwargs['ti']        total_value_string = ti.xcom_pull(task_ids='transform', key='total_order_value')        total_order_value = json.loads(total_value_string)
        print(total_order_value)
    # [END load_function]
    # [START main_flow]    extract_task = PythonOperator(        task_id='extract',        python_callable=extract,    )    extract_task.doc_md = """\#### Extract taskA simple Extract task to get data ready for the rest of the data pipeline.In this case, getting data is simulated by reading from a hardcoded JSON string.This data is then put into xcom, so that it can be processed by the next task."""
    transform_task = PythonOperator(        task_id='transform',        python_callable=transform,    )    transform_task.doc_md = """\#### Transform taskA simple Transform task which takes in the collection of order data from xcomand computes the total order value.This computed value is then put into xcom, so that it can be processed by the next task."""
    load_task = PythonOperator(        task_id='load',        python_callable=load,    )    load_task.doc_md = """\#### Load taskA simple Load task which takes in the result of the Transform task, by reading itfrom xcom and instead of saving it to end user review, just prints it out."""
    extract_task >> transform_task >> load_task
# [END main_flow]
# [END tutorial]

python ~/airflow/dags/tutorial.py

# 打印出所有正在活跃状态的 DAGsairflow list_dags
# 打印出 'tutorial' DAG 中所有的任务airflow list_tasks tutorial
# 打印出 'tutorial' DAG 的任务层次结构airflow list_tasks tutorial --tree

backfill            Run subsections of a DAG for a specified date rangelist_tasks          List the tasks within a DAGclear               Clear a set of task instance, as if they never ranpause               Pause a DAGunpause             Resume a paused DAGtrigger_dag         Trigger a DAG runpool                CRUD operations on poolsvariables           CRUD operations on variableskerberos            Start a kerberos ticket renewerrender              Render a task instance's template(s)run                 Run a single task instanceinitdb              Initialize the metadata databaselist_dags           List all the DAGsdag_state           Get the status of a dag runtask_failed_deps    Returns the unmet dependencies for a task instance                    from the perspective of the scheduler. In other words,                    why a task instance doesn't get scheduled and then                    queued by the scheduler, and then run by an executor).task_state          Get the status of a task instanceserve_logs          Serve logs generate by workertest                Test a task instance. This will run a task without                    checking for dependencies or recording it's state in                    the database.webserver           Start a Airflow webserver instanceresetdb             Burn down and rebuild the metadata databaseupgradedb           Upgrade the metadata database to latest versionscheduler           Start a scheduler instanceworker              Start a Celery worker nodeflower              Start a Celery Flowerversion             Show the versionconnections         List/Add/Delete connections