Bắt đầu ngayBắt đầu miễn phí

Định nghĩa DAG

Trong các bài trước, bạn đã hoàn thành riêng lẻ các bước extract, transform và load. Giờ đây, tất cả được gói gọn trong một hàm etl() gọn gàng mà bạn có thể xem trong console.

Hàm etl() trích xuất dữ liệu khóa học và đánh giá thô từ các cơ sở dữ liệu liên quan, làm sạch dữ liệu hỏng và điền giá trị bị thiếu, tính điểm đánh giá trung bình cho mỗi khóa học và tạo gợi ý dựa trên các quy tắc ra quyết định để sinh khuyến nghị, và cuối cùng nạp các khuyến nghị vào một cơ sở dữ liệu.

Như bạn có thể nhớ từ video, etl() nhận một đối số duy nhất: db_engines. Bạn có thể truyền đối số này cho task bằng op_kwargs trong PythonOperator. Bạn có thể truyền vào một dictionary, và nó sẽ được điền vào dưới dạng kwargs trong callable.

Bài tập này là một phần của khóa học

Introduction to Data Engineering

Xem khóa học

Hướng dẫn bài tập

  • Hoàn thiện định nghĩa DAG để chạy hằng ngày. Hãy dùng cú pháp cron.
  • Hoàn thiện PythonOperator() bằng cách truyền đúng các tham số. Ngoài etl, db_engines cũng đã có sẵn trong workspace của bạn.

Bài tập tương tác thực hành trực tiếp

Hãy thử làm bài tập này bằng cách hoàn thành đoạn mã mẫu này.

# Define the DAG so it runs on a daily basis
dag = DAG(dag_id="recommendations",
          schedule_interval="____")

# Make sure `etl()` is called in the operator. Pass the correct kwargs.
task_recommendations = PythonOperator(
    task_id="recommendations_task",
    python_callable=____,
    op_kwargs={"____": ____},
)
Chỉnh sửa và Chạy Mã