タスクをデコレートする
Airflow 上で Python を使った特定のタスクを実装しましょう。このタスクでは、pull_file 関数を使ってファイルをダウンロードし、Airflow 内のシステムに保存します。この関数はあらかじめ定義されています。
requests ライブラリと Airflow の dag オブジェクトはすでにインポートされています。
この演習はコースの一部です
Python で学ぶ Apache Airflow 入門
演習の手順
- Airflow ライブラリから必要なオブジェクトをインポートします。
pull_file関数を Airflow タスクとして設定します。- タスクオブジェクトを呼び出します。
実践的なインタラクティブ演習
このサンプルコードを完成させて、この演習に挑戦してみましょう。
# Import the task decorator
from ____ import ____
@dag(dag_id='file_update')
def file_update():
# Decorate the function as a task
____
def pull_file(URL, savepath):
r = requests.get(URL)
with open(savepath, 'wb') as f:
f.write(r.content)
print(f"File pulled from {URL} and saved to {savepath}")
# Call the task
____('http://dataserver/sales.json', 'latestsales.json')
file_update()