Apache Airflow คืออะไร?
เป็น Open Source Platform ตัวหนึ่งที่ออกแบบมาเพื่อการทำ schedule, monitor workflow และ data pipelines โดย Airflow มีการเขียน Workflow เป็น Directed Acyclic Graph(DAG) กราฟที่มีหัวลูกศรทิศทางเดียว ซึ่ง DAG ประกอบไปด้วยหลายๆ Task ที่เชื่อมต่อกันและในแต่ละ Task นั้นก็มีความสามารถที่แตกต่างกัน
ความสามารถของ Apache Airflow
1. Task scheduling: airflow สามารถตั้งเวลาในการทำงาน และสามารถ trigger จากภายนอกระบบได้
2. Dependency management: สามารถตั้งค่าลำดับการทำงานของ tasks ได้
3. Tasks execution: airflow รอบรับการทำงานหลาย platforms เช่น local machines, cloud environments (เช่น AWS, GCP), and containerized environments (เช่น Docker)
4. Monitoring and logging: airflow มีหน้าจอสำหรับการ monitor สถานะการทำงาน ของ workflows และสามารถตรวจสอบ logs การทำงานได้
5. Extensibility: เราสามารถสร้าง operators, sensors, and hooks สำหรับการใช้งานร่วมกับระบบภายนอก airflow ได้
หลักการของ Apache Airflow
1. Scalable: airflow มีสถาปัตยกรรมแบบโมดูลาร์ ที่สามารถปรับขนาด workers ได้ตามที่ต้องการ
2. Dynamic: เราสามารถสร้าง data pipelines ได้ตามต้องการด้วยภาษา python
3. Extensible: เราสามารถสร้าง operators ใหม่ๆจาก operators ของ airflow เพื่อให้ตรงกับความต้องการของงานแต่ละประเภทได้
4. Elegant: airflow มีการออกแบบให้ระบบมีความเรียบง่ายและยืดหยุ่นในการใช้งาน
ทำไมการทำ Airflow ถึงเป็นที่นิยม
⁃ มีฟังก์ชันการทำงานที่ครบครัน
⁃ สามารถที่จะจัดการ workflows ที่ซับซ้อนได้
⁃ ระบบมี UI ที่สามารถใช้งานง่าย
⁃ มีความยืดหยุ่นในการทำงานเพราะสามารถเขียนโค้ดในการสร้าง tasks ได้เอง
⁃ สามารถขยายขนาดของระบบ และ สร้าง operators จาก core operators ของ airflow ได้
⁃ สามารถ integrate กับ 3rd party ได้
⁃ เป็น open source ที่มี comunity ขนาดใหญ่
Example Scenario:
ลองนึกภาพสถานการณ์ที่เราจำเป็นต้องสร้าง workflow การประมวลผลข้อมูลรายวัน workflow เกี่ยวข้องกับสามงาน: task A task B และ task C
task A ดึงข้อมูลจากแหล่งที่มา task B ดำเนินการแปลงข้อมูล และ task C โหลดข้อมูลที่แปลงแล้วไปยังแหล่งเก็บข้อมูลปลายทาง
Setting up the DAG:
อย่างแรกเราต้องสร้าง Directed Acyclic Graph (DAG) ใน airflow
สร้าง python file ชื่อ ‘my_dag.py’ และ import modules ที่จำเป็น
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
ถัดมาสร้าง DAG พร้อมกับ default arguments:
default_args = {
‘owner’: ‘airflow’,
‘start_date’: datetime(2023, 1, 1),
}
dag = DAG(‘daily_data_workflow’, default_args=default_args, schedule_interval=’@daily’)
Defining Tasks:
ตอนนี้เราจะสร้าง tasks ขึ้นมา 3 tasks ด้วย python functions และ สร้าง operators ที่เกี่ยวข้องของแต่ละ task:
# Task A
def task_a_function():
print(“Executing Task A – Data Retrieval”)
task_a = PythonOperator(
task_id=’task_a’,
python_callable=task_a_function,
dag=dag
)
# Task B
def task_b_function():
print(“Executing Task B – Data Transformation”)
task_b = PythonOperator(
task_id=’task_b’,
python_callable=task_b_function,
dag=dag
)
# Task C
def task_c_function():
print(“Executing Task C – Data Loading”)
task_c = PythonOperator(
task_id=’task_c’,
python_callable=task_c_function,
dag=dag
)
Task Dependencies:
เพื่อบอกลำดับการทำงานระหว่าง tasks โดยใช้ >> operator:
task_a >> task_b >> task_c
ในตัวอย่างนี้ task A เป็นข้อกำหนดเบื้องต้นสำหรับ task B และ task B เป็นข้อกำหนดเบื้องต้นสำหรับ task C ซึ่งหมายความว่า task C จะดำเนินการต่อเมื่อ task A และ task B เสร็จสมบูรณ์แล้วเท่านั้น
Running the Workflow:
ในการเรียกใช้ workflow Scheduler ของ Airflow จะดึง DAG ขึ้นมาทำงานตามช่วงเวลาที่ระบุ (@daily ในกรณีของเราจะทำงานทุกวัน)
เมื่อ save file DAG แล้วเข้าไปที่ Airflow จะเห็น DAG ที่เราสร้างขึ้นมาบอกสถานะการทำงาน และ schedule ที่ตั้งไว้
การติดตั้ง Apache Airflow ด้วย Docker
Step 1: Fetching docker-compose.yaml
สร้าง directory airflow ของเราขึ้นมา แล้ว download docker-compose.yaml ด้วยคำสั่งข้างล่างนี้
mkdir airflow-setup curl –LfO ‘https://airflow.apache.org/docs/apache-airflow/2.6.1/docker-compose.yaml‘ |
ใน docker-compose.yaml จะประกอบไปด้วย services ดังนี้
• airflow-scheduler – scheduler จะตรวจสอบ tasks และ DAG ทั้งหมดจากนั้นจะ trigger task ถัดไปเมื่อ งานก่อนหน้าทำงานสำเร็จ
• airflow-worker – worker ที่ประมวลผล tasks ที่ถูกป้อนให้โดยscheduler
• airflow-triggerer – The triggerer runs an event loop for deferrable tasks.
• airflow-init – initialization service
• postgres – database
• redis – ตัวกลางที่คอยป้อนงานจาก scheduler ให้กับ worker
Step 2: Initializing Environment
ก่อนการรัน airflow ครั้งแรกต้องสร้าง directories ที่จำเป็น และ initialize database
Directories ที่จำเป็น
• ./dags – you can put your DAG files here.
• ./logs – contains logs from task execution and scheduler.
• ./config – you can add custom log parser or add airflow_local_settings.py to configure cluster policy.
สร้าง directories ที่จำเป็นด้วยคำสั่ง
mkdir –p ./dags ./logs ./plugins ./config |
Initialize database
docker compose up airflow-init |
หลังจาก initialize สำเร็จจะได้ผลลัพธ์หน้าตาแบบนี้
airflow-init_1 | Upgrades done |
จะมี account ถูกสร้างขึ้นมา สามารถ login ด้วย username airflow และpassword airflow
Step 3: Running Airflow
ตอนนี้สามารถ start airflow ได้แล้วด้วยคำสั่ง
docker compose up |
แล้วเราสามารถ login เข้าใช้งาน web ui ของ airflow ได้ที่http://localhost:808
