Dynamic Task Mapping

Xây dựng Data Pipeline với Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Vấn đề với các task hardcode

@task
def process_file_1():
    transform("/data/file_1.csv")
@task
def process_file_2():
    transform("/data/file_2.csv")
@task
def process_file_3():
    transform("/data/file_3.csv")

Ví dụ Dynamic Task Mapping

.expand()

@task
def fetch_files():
    return ["/data/file_1.csv", "/data/file_2.csv", "/data/file_3.csv"]

@task(max_active_tis_per_dagrun=2)  # kiểm soát mức song song tối đa
def process_file(path: str):
    print(f"processing file: {path}")


files = fetch_files()
process_file.expand(path=files)

Mở rộng khi chạy

Các task được map trong UI

Các instance được map hiển thị dưới dạng instance có chỉ số trong một task duy nhất
Nhấp vào từng cái để xem log và trạng thái riêng
Dag vẫn dễ đọc ngay cả với hàng trăm instance

Các task được map trong UI

.partial()

@task
def process_file(path: str, output_dir: str):
    transform(path, output_dir)


files = get_files()
process_file.partial(output_dir="/out").expand(path=files)

Vấn đề tích Descartes (cross-product)

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Tích Descartes: 3 x 3 = 9 instance (không như mong muốn!)
process.expand(path=files, dest=destinations)

Ghép list theo cặp

.zip()

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Ghép cặp: 3 instance (a->a, b->b, c->c)
process.expand_kwargs(files.zip(destinations))

zip() ghép phần tử một-một theo vị trí
expand_kwargs bung từng cặp thành đối số keyword

Dùng cái nào khi nào

Mẫu	Trường hợp dùng
`expand()`	Map qua một danh sách
`partial()`	Cố định tham số dùng chung giữa các instance
`zip()`	Ghép cặp nhiều danh sách theo vị trí

Giúp xây dựng Dag có khả năng mở rộng
Khi dữ liệu thay đổi, số lượng task tự điều chỉnh khi chạy

Cùng luyện tập nào!

Xây dựng Data Pipeline với Airflow