https://mpolinowski.github.io/assets/js/8cce1bed.b76a583f.js

1"use strict";(globalThis.webpackChunkmikes_dev_notebook=globalThis.webpackChunkmikes_dev_notebook||[]).push([[10622],{234156(e,a,t){t.r(a),t.d(a,{assets:()=>d,contentTitle:()=>o,default:()=>h,frontMatter:()=>i,metadata:()=>n,toc:()=>l});const n=JSON.parse('{"id":"IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler/index","title":"Apache Airflow DAG Scheduling","description":"Airflow is a platform to author, schedule and monitor workflows.","source":"@site/docs/IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler/index.md","sourceDirName":"IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler","slug":"/IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler/2023-02-05","permalink":"/docs/IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler/2023-02-05","draft":false,"unlisted":false,"editUrl":"https://github.com/facebook/docusaurus/tree/main/packages/create-docusaurus/templates/shared/docs/IoT-and-Machine-Learning/AIOps/2023-02-05-apache-airflow-scheduler/index.md","tags":[{"inline":true,"label":"Python","permalink":"/docs/tags/python"},{"inline":true,"label":"Machine Learning","permalink":"/docs/tags/machine-learning"},{"inline":true,"label":"Airflow","permalink":"/docs/tags/airflow"}],"version":"current","sidebarPosition":4690,"frontMatter":{"sidebar_position":4690,"slug":"2023-02-05","title":"Apache Airflow DAG Scheduling","authors":"mpolinowski","tags":["Python","Machine Learning","Airflow"],"description":"Airflow is a platform to author, schedule and monitor workflows."},"sidebar":"tutorialSidebar","previous":{"title":"Apache Airflow Dynamic DAGs","permalink":"/docs/IoT-and-Machine-Learning/AIOps/2023-02-06-apache-airflow-dynamic-dags/2023-02-06"},"next":{"title":"Apache Airflow Data Pipelines","permalink":"/docs/IoT-and-Machine-Learning/AIOps/2023-02-04-apache-airflow-data-pipelines/2023-02-04"}}');var s=t(474848),r=t(28453);const i={sidebar_position:4690,slug:"2023-02-05",title:"Apache Airflow DAG Scheduling",authors:"mpolinowski",tags:["Python","Machine Learning","Airflow"],description:"Airflow is a platform to author, schedule and monitor workflows."},o=void 0,d={},l=[{value:"Basic Scheduling",id:"basic-scheduling",level:2},{value:"Use Dataset Updates as Worflow Trigger",id:"use-dataset-updates-as-worflow-trigger",level:2},{value:"Debugging DAGs",id:"debugging-dags",level:3},{value:"Importing Datasets",id:"importing-datasets",level:3},{value:"Airflow Sensors",id:"airflow-sensors",level:2}];function c(e){const a={a:"a",code:"code",h2:"h2",h3:"h3",img:"img",li:"li",p:"p",pre:"pre",strong:"strong",ul:"ul",...(0,r.R)(),...e.components};return(0,s.jsxs)(s.Fragment,{children:[(0,s.jsx)(a.p,{children:(0,s.jsx)(a.img,{alt:"Guangzhou, China",src:t(299187).A+"",width:"1061",height:"405"})}),"\n",(0,s.jsxs)(a.ul,{children:["\n",(0,s.jsx)(a.li,{children:(0,s.jsx)(a.a,{href:"#basic-scheduling",children:"Basic Scheduling"})}),"\n",(0,s.jsxs)(a.li,{children:[(0,s.jsx)(a.a,{href:"#use-dataset-updates-as-worflow-trigger",children:"Use Dataset Updates as Worflow Trigger"}),"\n",(0,s.jsxs)(a.ul,{children:["\n",(0,s.jsx)(a.li,{children:(0,s.jsx)(a.a,{href:"#debugging-dags",children:"Debugging DAGs"})}),"\n",(0,s.jsx)(a.li,{children:(0,s.jsx)(a.a,{href:"#importing-datasets",children:"Importing Datasets"})}),"\n"]}),"\n"]}),"\n",(0,s.jsx)(a.li,{children:(0,s.jsx)(a.a,{href:"#airflow-sensors",children:"Airflow Sensors"})}),"\n"]}),"\n",(0,s.jsxs)(a.ul,{children:["\n",(0,s.jsx)(a.li,{children:(0,s.jsx)(a.a,{href:"https://github.com/mpolinowski/apache-airflow-intro",children:"Github Repository"})}),"\n"]}),"\n",(0,s.jsx)(a.h2,{id:"basic-scheduling",children:"Basic Scheduling"}),"\n",(0,s.jsxs)(a.p,{children:["As we have seen before every pipeline we build is defined with a ",(0,s.jsx)(a.code,{children:"start_date"})," and a scheduled interval in  which the pipeline should be run. The ",(0,s.jsx)(a.code,{children:"catchup"})," parameter tells the DAG if it should try to run all the non-triggered DAG-runs since the start. Another, optional parameter is ",(0,s.jsx)(a.code,{children:"end_date"}),":"]}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:'with DAG("example_dag", start_date=datetime(2023,2,5), schedule="@daily", cat

1chup=False):\n'})}),"\n",(0,s.jsx)(a.h2,{id:"use-dataset-updates-as-worflow-trigger",children:"Use Dataset Updates as Worflow Trigger"}),"\n",(0,s.jsxs)(a.p,{children:["You can use ",(0,s.jsx)(a.strong,{children:"Dataset Scheduler"})," e.g. when you are working with multiple Airflow DAGs that depend on the results from the previous pipeline. So that pipeline 2 is automatically triggered once pipeline 1 finished updating a dataset."]}),"\n",(0,s.jsx)(a.p,{children:"So we first write a DAG that pre-processes our dataset - simulated by adding some text to a simple text file:"}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:"from airflow import DAG, Dataset\nfrom airflow.decorators import task\n\nfrom datetime import datetime\n\ndataset = Dataset('/tmp/preprocessed_data.txt') \nstart_date = datetime(2023,2,5)\ndescription = \"Data pre-processing\"\ntags = ['scheduler', 'producer']\ndag_id = 'producer'\nschedule = '@daily'\n\nwith DAG(\n    dag_id=dag_id,\n    start_date=start_date,\n    schedule=schedule,\n    catchup=False,\n    description=description,\n    tags=tags\n    ):\n\n    ## decorator with outlet so updates\n    ## to dataset can be used as trigger\n    @task(outlets=[dataset])\n    ## preprocess data and write to file\n    def preprocess_dataset():\n        with open(dataset.uri, 'a+') as file:\n            file.write('preprocessed data ready for consumption')\n\n    preprocess_dataset()\n"})}),"\n",(0,s.jsxs)(a.p,{children:["The ",(0,s.jsx)(a.code,{children:"@task"})," decorator now has an ",(0,s.jsx)(a.code,{children:"outlet"})," that tells Airflow that once this task ran the dataset ",(0,s.jsx)(a.code,{children:"dataset"})," was updated and can use this event to trigger the following DAG:"]}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:"from airflow import DAG, Dataset\nfrom airflow.decorators import task\n\nfrom datetime import datetime\n\ndataset = Dataset('/tmp/preprocessed_data.txt') \nstart_date = datetime(2023,2,5)\ndescription = \"Start this pipeline when new data arrives\"\ntags = ['scheduler', 'consumer']\ndag_id = 'consumer'\n\nwith DAG(\n    start_date=start_date,\n    dag_id=dag_id,\n    schedule=[dataset],\n    catchup=False,\n    description=description,\n    tags=tags\n    ):\n\n    @task\n    ## process prepared data and write to file\n    def process_dataset():\n        with open(dataset.uri, 'r') as file:\n            print(file.read())\n\n    process_dataset()\n"})}),"\n",(0,s.jsx)(a.p,{children:(0,s.jsx)(a.img,{alt:"Apache Airflow DAG Scheduling",src:t(20978).A+"",width:"1331",height:"374"})}),"\n",(0,s.jsxs)(a.p,{children:["Now as soon as we trigger the ",(0,s.jsx)(a.strong,{children:"Producer DAG"})," the dataset in form of the text file will be updated which in turn triggers the ",(0,s.jsx)(a.strong,{children:"Consumer DAG"}),":"]}),"\n",(0,s.jsx)(a.p,{children:(0,s.jsx)(a.img,{alt:"Apache Airflow DAG Scheduling",src:t(835593).A+"",width:"1337",height:"372"})}),"\n",(0,s.jsx)(a.h3,{id:"debugging-dags",children:"Debugging DAGs"}),"\n",(0,s.jsx)(a.p,{children:"I am certain that I was able to find the execution logs in the Airflow UI before... hmm not sure where they are hiding now. But you can enter the Airflow container and get the task log from there:"}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-bash",children:"docker exec -ti airflow /bin/bash\n"})}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-bash",children:"airflow info\n\nApache Airflow\nversion                | 2.5.1                                              \nexecutor               | SequentialExecutor                                 \ntask_logging_handler   | airflow.utils.log.file_task_handler.FileTaskHandler\nsql_alchemy_conn       | sqlite:////opt/airflow/airflow.db                  \ndags_folder            | /opt/airflow/dags                                  \nplugins_folder         | /opt/airflow/plugins                               \nbase_log_folder        | /opt/airflow/logs\n\n...\n"})}),"\n",(0,s.jsx)(a.p,{children:"I was able to find the log file in:"}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-bash",children:"/opt/airflow/logs/dag_id=producer/run_id=manual__2023-02-05T13:10:07.002907+00:00/task_id=preprocess_dataset\n"})}),"\n",(0,s.jsx)(a.p,{children:"And catch the Python error that was crashing my flow."}),"\n",(0,s.jsx)(a.h3,{id:"importing-datasets",children:"Importing Datasets"}),"\n",(0,s.jsx)(a.p,{children:"In the example above we needed to define the dataset in both DAGs. A cleaner solution is to have a dataset definition file instead:"}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-bash",children:"mkdir /opt/airflow/dags/include\nnano /opt/airflow/dags/include/datasets.py\n"})}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:"from airflow import Dataset\n\nDATASET_01 = Dataset('/tmp/preprocessed_data.txt') \n"})}),"\n",(0,s.jsx)(a.p,{children:"Now we can import the dataset where ever we need it:"}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:"from airflow import DAG\nfrom airflow.decorators import task\n\nfrom datetime import datetime\n\nfrom include.datasets import DATASET_01\n\n\nstart_date = datetime(2023,2,5)\ndescription = \"Data pre-processing\"\ntags = ['scheduler', 'producer']\ndag_id = 'producer'\nschedule = '@daily'\n\nwith DAG(\n    dag_id=dag_id,\n    start_date=start_date,\n    schedule=schedule,\n    catchup=False,\n    description=description,\n    tags=tags\n    ):\n\n    ## decorator with outlet so updates\n    ## to dataset can be used as trigger\n    @task(outlets=[DATASET_01])\n    ## preprocess data and write to file\n    def preprocess_dataset():\n        with open(DATASET_01.uri, 'a+') as file:\n            file.write('preprocessed data ready for consumption')\n\n    preprocess_dataset()\n"})}),"\n",(0,s.jsxs)(a.p,{children:["All your datasets and DAGs that depend on it can be monitored under the ",(0,s.jsx)(a.strong,{children:"Datasets"})," tab:"]}),"\n",(0,s.jsx)(a.p,{children:(0,s.jsx)(a.img,{alt:"Apache Airflow DAG Scheduling",src:t(957472).A+"",width:"1075",height:"301"})}),"\n",(0,s.jsx)(a.h2,{id:"airflow-sensors",children:"Airflow Sensors"}),"\n",(0,s.jsxs)(a.p,{children:["With ",(0,s.jsx)(a.strong,{children:"Dataset Schedulers"})," Airflow monitors the successful completion of a task to trigger the next DAG. If we want to be able to react to other programs updating datasets we need to use ",(0,s.jsx)(a.a,{href:"https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html",children:"Airflow Sensors"})," instead. Sensors are a special type of Operator that are designed to do exactly one thing - wait for something to occur. It can be time-based, or waiting for a file, or an external event, but all they do is wait until something happens, and then succeed so their downstream tasks can run."]}),"\n",(0,s.jsxs)(a.p,{children:["In the following example we receive data from 3 actors, have to pre-process it and ",(0,s.jsx)(a.strong,{children:"after all data is collected"})," process the prepared data:"]}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-bash",children:"Data Producer A -> Pre-Processing ->\nData Producer B -> Pre-Processing -> Processing\nData Producer C -> Pre-Processing ->\n"})}),"\n",(0,s.jsx)(a.pre,{children:(0,s.jsx)(a.code,{className:"language-py",children:"from airflow.models import DAG\nfrom airflow.sensors.filesystem import FileSensor\n\nfrom datetime import datetime\n\nstart_date = datetime(2023,2,5)\ndescription = \"Data pre-processing\"\ntags = ['sensor']\nschedule = '@daily'\n\n\nwith DAG('dag_sensor',\n    schedule=schedule,\n    start_date=start_date,\n    description=description,\n    tags=tags\n    catchup=False\n    ):\n\n    sensor_task = FileSensor(\n        task_id='waiting_for_change',\n        poke_interval=30,\n        # kill sensor if data does\n        # not arrive\n        timeout= 60 * 5,\n        # 'reschedule' or 'poke'\n        # the first releases the\n        # worker slot in between pokes\n        mode='reschedule',\n        # skip sensor after timeout\n        soft_fail=True,\n        filepath= '/tmp/input_data.txt'\n    )\n\n    ...\n"})})]})}function h(e={}){const{wrapper:a}={...(0,r.R)(),...e.components};return a?(0,s.jsx)(a,{...e,children:(0,s.jsx)(c,{...e})}):c(e)}},20978(e,a,t){const n=t.p+"assets/images/Apache_Airflow_Scheduler_01-9b2336eb794e1616c3f526d14392b818.png";t.d(a,["A",0,n])},835593(e,a,t){const n=t.p+"assets/images/Apache_Airflow_Scheduler_02-b8dee20bc96860bbe696a128d0c20114.png";t.d(a,["A",0,n])},957472(e,a,t){const n=t.p+"assets/images/Apache_Airflow_Scheduler_03-98baca4db3fa4aeb786fd58ec9eb7278.png";t.d(a,["A",0,n])},299187(e,a,t){const n=t.p+"assets/images/photo-kt443t6d_64hdh43hfh6dgjdfhg4_d-fe9bbb57ea8da08fea2f3fef2bf2515b.jpg";t.d(a,["A",0,n])},28453(e,a,t){t.d(a,{R:()=>i,x:()=>o});var n=t(296540);const s={},r=n.createContext(s);function i(e){const a=n.useContext(r);return n.useMemo(function(){return"function"==typeof e?e(a):{...a,...e}},[a,e])}function o(e){let a;return a=e.disableParentContext?"function"==typeof e.components?e.components(s):e.components||s:i(e.components),n.createElement(r.Provider,{value:a},e.children)}}}]);