Airflow에서 매개 변수를 사용하는 예는 무엇입니까?

Question

자주 다운로드하는 파일의 파일 이름에 날짜가 있습니다.

csat_surveys_2017_03_05.csv 
03062017_roster.csv 

내 코드는이 코드를 개별적으로 처리합니다.

• 파일 이름을 추가 존재한다 예상되는 날짜와 (파일 이름 슬라이스 기준) 처리 된 파일 목록에서 날짜 (일부 날짜가 현재 날짜까지 범위) 내가 처리 할 각 파일에 대해
• 비교 해당 테이블에 추가되지 않은 새 파일 만 처리합니다.

이 논리를 코딩해야 할 필요를 대체하기 위해 기류 일정 날짜를 사용할 수 있습니까? 매일 작업이 실행되도록 예약됩니다. 필자는 예정된 날짜 (아마도 1 일 빼기)를 가져 와서 해당 값을 매개 변수로 사용하여 파일 이름의 일부로 전달합니다 (팬더에서). 그렇다면 템플릿으로 사용할 수있는 명확한 예를 참조 할 수 있습니까?

파일을 놓치거나 며칠 동안 지연되는 경우 더 나은 접근 방법입니다 (필자는 작업이 실패하기를 원하고, 성공할 때까지 계속 노력하고 있거나 고객에게 문제 제기)?

Answer 1

예라고 말하고 싶지만 execution_date을 사용하는 것이 좋습니다.

액세스하려면 템플릿 입력란이 필요합니다.

my_task = MyOperator(
    task_id='t1', 
    filename='prefix_{{ ds }}_suffix') 

ds입니다 : 일부 기본 운영자는 이미 사람들이 있거나 다음이 같은 모양 자신의 연산자 만들 수 있습니다 : 귀하의 DAG에서

을, 당신은 작업이있을 것이다 execution_date 매개 변수를 날짜의 문자열 표현으로 액세스하기위한 기류 매크로.

그리고 당신의 MyOperator과 같을 것이다 :

class MyOperator(BaseOperator): 
    template_fields = ('filename') 

    def __init__(self, filename) 
     self.filename = filename 

    def execute(self, context): 
     download_file(self.filename) 
     do_other_stuff() 

당신은 매크로 섹션에서 작업을 변수화하는 방법에 대한 자세한 내용을 확인할 수 있습니다

https://airflow.incubator.apache.org/code.html#macros