气流 - 无法将Bashoperator中的模板Env变量推到XCOM

Question

我正在尝试对XCOM和Jinja模板上的云作曲家的气流2.2基础知识感到满意。我的Bashoperator的代码如下（它以文件的名称使用执行日期）。该文件名作为env变量传递，以便{{ts}}被识别为模板（如果我将其作为params变量传递给，执行日期不会被解析为params未根据源代码）：

    writeToFile1Bash = BashOperator (
        task_id='writetofile1bash',
        retries=0,
        bash_command='echo "number,square,root\n" > ${file1name} && echo "{{ task_instance.xcom_push(key="file1name", value=$file1name) }}"',
        env={
            'file1name':'/home/airflow/gcs/data/file1-{{ ts }}.txt'
        }
    )

对变量$ {file1name}的第一个引用是按预期工作的，如果我完成了bash命令在那里，它将成功创建文件。但是，我希望能够将其作为XCOM（带有自定义变量名称）传递到下游任务，而我无法实现它。我在第二个命令中提取file1name变量的不同格式以下错误：

- var.value.file.file1name - ＆gt; keyError：'变量file1name不存在'

- $（file1name）或$ file1name - ＆gt; 错误渲染模板：101

- $ {file1name} - ＆gt; jinja2.exceptions.templatesyntaxerror：预期令牌'：'，get'}'}'

- env.file1name，env> env ['file1name']或<代码> env ['file1name] - ＆gt; env未定义

- params.file1name1 - ＆gt; 错误渲染模板：类型strictundefined的对象不是JSON序列化

我尝试了每种想法的方式，但是文档尚不清楚。即使在此 tutorial （短语之后：之后：修改DAG后，您应该拥有以下操作：），鼓励您使用env> env而不是params，因为需要jinja-parssparible，但是;没有一个代码示例，即如何从双括号内部引用此变量{{}}。

如果我已经花了很多时间，我会非常感谢是否有人可以帮助解决这个问题。

Answer 1

对于bashoperator，stdout的最后一行将推到XCOM。

参考： https://airflow.apache.org/docs/apache-airflow/1.10.13/_api/_api/airflow/airflow/operators/bash_operator/index.html#airflow.operators.bash_operator.bashoperator.bashoperator.bashoperator.bashoperator

您可以做这样：

writeToFile1Bash = BashOperator (
    task_id='writetofile1bash',
    retries=0,
    bash_command='echo "number,square,root\n" > ${file1name} && echo $file1name',
    env={
        'file1name':'/home/airflow/gcs/data/file1-{{ ts }}.txt'
    }
)

然后在连续的运算符中：

puller = BashOperator(
    task_id='puller',
    bash_command='echo {{ ti.xcom_pull(task_ids="writetofile1bash")}}'
)

update

否XCOM方法：

{{{ts}}是dag_run的Timestamp SO TS不会更改在任务之间。

也就是说，在同一dag中多次使用{{ts}}是安全的，并且使用静态变量，您不必使用XCOM。

filename = '/home/airflow/gcs/data/file1-{{ ts }}.txt'

writeToFile1Bash = BashOperator (
    task_id='writetofile1bash',
    retries=0,
    bash_command='echo "number,square,root\n" > ${file1name} && echo $file1name',
    env={
        'file1name': filename
    }
)

puller = BashOperator(
    task_id='puller',
    bash_command=f'echo {filename}'
)