@task 더 활용하기
워크플로우 구현을 이어가려면, 다운로드한 파일의 변경 사항을 파싱하고 저장하는 단계를 추가해야 합니다. 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()