Operator trong Airflow

Giới thiệu về Apache Airflow bằng Python

Mike Metzger

Data Engineer

Operator

Đại diện một tác vụ trong luồng công việc
Thường chạy độc lập
Thường không chia sẻ thông tin
Nhiều operator cho các loại tác vụ khác nhau

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

  task1()

example_dag()

@task (PythonOperator)

Thực thi một hàm Python
Bất kỳ hàm Python nào cũng có thể dùng @task
Có thể truyền dữ liệu giữa các hàm/tác vụ

from airflow.sdk import task

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

printme()

Tham số cho @task

Có thể truyền tham số cho tác vụ/hàm như hàm Python thông thường

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


printme(name='DataCamp')


# Thêm vào log:
#   Hi DataCamp - This goes in the logs!
# trong Airflow

@task.bash (BashOperator)

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

bash_example()

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

run_cleanup()

Chạy lệnh/script Bash được chỉ định
Chạy lệnh trong thư mục tạm
Có thể chỉ định biến môi trường cho lệnh

Phụ thuộc giữa tác vụ

Mỗi DAG có một tập tác vụ cần hoàn thành
Phụ thuộc tác vụ xác định thứ tự chạy
Có nhiều cách khai báo phụ thuộc

DAG Airflow gồm các tác vụ nối nhau thể hiện thứ tự phụ thuộc

Cú pháp bitshift

>> và <<

task1() >> task2()

# task1 hoàn tất trước khi task2 bắt đầu


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

# task1 xong trước task2 và task2 xong trước task3


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

# task1 và task2 có thể chạy song song nhưng cả hai phải xong trước task3

Ví dụ cú pháp bitshift

# Tải dữ liệu bán hàng trước khi đối soát
download_sales_data() >> reconcile()

# Tải dữ liệu tồn kho trước khi đối soát
download_inventory_data() >> reconcile()

Cả hai phải hoàn tất trước khi đối soát, nhưng không bắt buộc thứ tự

Ayo berlatih!

Giới thiệu về Apache Airflow bằng Python