dag.py

one_day_ago = datetime.combine(datetime.today() - timedelta(1), datetime.min.time())

default_args = {
    'owner': 'airflow',
    'start_date': one_day_ago
}

dag = DAG(dag_id='my_pipeline', default_args=default_args)

generate_build_id = BuildIdOperator(dag=dag)
# What is the right way to pass input_path, output_path as parameters
# when starting DAG?
scalding_operator = ScaldingOperator(input_path="{{ dag_run.conf['input_path'] }}",
                                     output_path="{{ dag_run.conf['output_path'] }}",
                                     dag=dag, task_id='my-job-scalding', 
                                     artifact="com.company:my-job:{{ dag_run.conf['my_job_version'] }}")

scalding_operator.set_upstream(generate_build_id)