更多 @tasks

为继续实现您的工作流，您需要再添加一步，用于解析并保存已下载文件的更改。Dag process_sales 已定义，并且已添加了 pull_file 任务。本题中，Python 函数也已为您定义好：parse_file(inputfile, outputfile)。

请注意，在实现 Airflow 任务时，您不一定总能理解给定的每个具体步骤。只要您能理解如何将这些步骤封装进 Airflow 的结构中，您就能够实现所需的工作流。

本练习是课程的一部分

Python 中的 Apache Airflow 入门

查看课程

练习说明

使用 parse_file 方法创建一个 Airflow 任务。
使用必要的参数调用该任务。

交互式实操练习

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

@dag(dag_id='process_sales')
def process_sales():
    # Decorate parse_file as a task
    ____
    def parse_file(inputfile: str, outputfile: str):
        with open(inputfile) as infile:
            data = json.load(infile)
            with open(outputfile, 'w') as outfile:
                json.dump(data, outfile)

    pull_file('http://dataserver/sales.json', 'latestsales.json')
    # Call the parse_file task
    ____('latestsales.json', 'latestsales_parsed.json')

process_sales()

编辑并运行代码