开始使用免费开始使用

定义 DAG

在前面的练习中，您已经分别完成了抽取、转换和加载三个阶段。现在，这些步骤被整合到一个简洁的 etl() 函数中，您可以在控制台中查看。

etl() 函数会从相关数据库中抽取原始课程与评分数据，清洗损坏数据并填补缺失值，计算每门课程的平均评分，并依据给定的决策规则生成推荐，最后将这些推荐结果加载到数据库中。

正如您在视频中看到的，etl() 接受一个参数：db_engines。您可以在 PythonOperator 中通过 op_kwargs 将其传给任务。可以传入一个字典，该字典会作为可调用对象的 kwargs 进行填充。

本练习是课程的一部分

Data Engineering 入门

查看课程

练习说明

  • 完成 DAG 定义，使其按天运行。请使用 cron 表达式。
  • 补全 PythonOperator()，传入正确的参数。除 etl 外，db_engines 也已在您的工作区中可用。

交互式实操练习

通过完成这段示例代码来试试这个练习。

# 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={"____": ____},
)
编辑并运行代码