Airflow operators

Apache Airflow เบื้องต้นด้วย Python

Mike Metzger

Data Engineer

Operators

แทนงานเดียวในเวิร์กโฟลว์
ทำงานอิสระจากกัน (โดยทั่วไป)
โดยทั่วไปไม่แชร์ข้อมูลกัน
มี operator หลายประเภทสำหรับงานต่างกัน

@dag(
  dag_id="Example_Dag"
)
def example_dag():
  @task
  def task1():
    return "The result from task1"

  task1()

example_dag()

@task (PythonOperator)

รันฟังก์ชัน Python
ฟังก์ชัน Python ใดก็ได้สามารถใช้ decorator @task ได้
ส่งผ่านข้อมูลระหว่างฟังก์ชัน / task ได้

from airflow.sdk import task

@task
def printme():
    print("This goes in the logs!")

printme()

อาร์กิวเมนต์ของ @task

ส่งอาร์กิวเมนต์ให้ task / ฟังก์ชันได้เหมือนฟังก์ชัน Python ทั่วไป

@task
def printme(name: str):
    print(f"Hi {name} - This goes in the logs!")


printme(name='DataCamp')


# Adds:
#   Hi DataCamp - This goes in the logs!
# to the Airflow logs

@task.bash (BashOperator)

@task.bash
def bash_example():
  return "echo 'Example!'"

bash_example()

@task.bash
def run_cleanup():
  return "runcleanup.sh"

run_cleanup()

รันคำสั่งหรือสคริปต์ Bash ที่กำหนด
รันคำสั่งในไดเรกทอรีชั่วคราว
กำหนด environment variable สำหรับคำสั่งได้

Task dependencies

แต่ละ DAG มีชุด task ที่ต้องทำให้เสร็จ
dependency ของ task กำหนดลำดับการรัน
มีวิธีระบุ dependency ได้หลายแบบ

Airflow DAG ที่แสดง task เชื่อมต่อกันพร้อมลำดับ dependency

Bitshift syntax

>> และ <<

task1() >> task2()

# task1 completes before task2 starts


task1() >> task2() >> task3()

# task1 completes before task2 and task2 completes before task3


task1() >> task3()
task2() >> task3()

# task1 and task2 can run together but both must complete before task3 runs

ตัวอย่าง Bitshift syntax

# Download sales data before reconciling
download_sales_data() >> reconcile()

# Download inventory data before reconciling
download_inventory_data() >> reconcile()

ทั้งสองต้องเสร็จก่อน reconcile แต่ไม่กำหนดลำดับระหว่างกัน

มาฝึกกันเถอะ!

Apache Airflow เบื้องต้นด้วย Python