from datetime import timedelta, datetime
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from airflow.operators.bash_operator import BashOperator
from airflow.models import Variable
from airflow.utils.dates import days_ago
args1 = {
'owner': 'Airflow',
'start_date': days_ago(2),
}
dag = DAG(
dag_id='anantha-dag',
default_args=args1,
schedule_interval='0 0 * * *',
dagrun_timeout=timedelta(minutes=60),
tags=['ananthaPOC']
)
spark_submit_operator = SparkSubmitOperator(
task_id='spark_submit_job',
dag=dag,
application='/opt/cloudera/parcels/SPARK2/lib/spark2/examples/jars/spark-examples_2.11-2.4.0.cloudera2.jar',
application_args=[ '10' ],
spark_binary="{{ dag_run.conf["spark_binary_1"] if dag_run else "" }}",
java_class="org.apache.spark.examples.SparkPi",
**kwargs
)
bash_task = BashOperator(
task_id="bash_task",
bash_command='echo "Here is the message: \'{{ dag_run.conf["spark_binary_1"] if dag_run else "" }}\'"',
dag=dag,
)
bash_task >> spark_submit_operator
我有一个示例dag和jinja模板在bash操作符中工作,但在spark submit操作符中没有。下面是错误:
spark_binary={{ dag_run.conf["spark_binary_1"] if dag_run else "" }},
NameError: name 'dag_run' is not defined
我正在通过cli命令调用dag:aiffolTrigger\u dag anantha dag--conf“{{'spark\u binary\u 1':'spark2 submit'}”
暂无答案!
目前还没有任何答案,快来回答吧!