如何将参数传递给 PapermillOperator 以在气流上运行作业？答案

【问题标题】：how to pass parameters to PapermillOperator to run job on airflow?如何将参数传递给 PapermillOperator 以在气流上运行作业？
【发布时间】：2019-10-31 06:22:33
【问题描述】：

使用 PapermillOperator 运行气流作业时，dag 执行失败。

我在将参数传递给 PapermillOperator 时遇到问题。

我打开 papermill_operator.py (packages/airflow/operators/papermill_operator.py) 并硬编码一行来指定 paparameters

def execute(self, context):
        for i in range(len(self.inlets)):
            pm.execute_notebook(self.inlets[i].location, 
                                self.outlets[i].location,
                                parameters = dict(msgs="hello")
                                progress_bar=False, report_mode=True)

然后就可以了

而原来的代码是

def execute(self, context):
        for i in range(len(self.inlets)):
            pm.execute_notebook(self.inlets[i].location, 
                                self.outlets[i].location,
                                parameters=self.inlets[i].parameters,
                                progress_bar=False, report_mode=True)

尝试了另一种解决方案 https://github.com/nteract/papermill/issues/324#issuecomment-472446375 它工作正常

我的 DAG 代码是

import airflow

from airflow.models import DAG
from airflow.operators.papermill_operator import PapermillOperator

from datetime import timedelta

args = {
    'owner': 'Airflow',
    'start_date': airflow.utils.dates.days_ago(2),

}

dag = DAG(
    dag_id='9', default_args=args,
    schedule_interval='@once',
    dagrun_timeout=timedelta(minutes=10))

run_this = PapermillOperator(
    task_id="1",
    dag=dag,
    input_nb="/home/exa00112/abc.ipynb",
    output_nb="/home/exa00112/umesh.ipynb",
    parameters = dict("msgs" = "hello")
)

run_this

[2019-09-10 20:36:48,806] {logging_mixin.py:95} INFO - [2019-09-10 > > > 20:36:48,806] {datasets.py:62} INFO - 参数 [2019-09-10 20:36:48,806] {init.py:1580} 错误 - 无法编译非 > 模板节点回溯（最近一次通话最后）： _run_raw_task 中的文件“/usr/local/lib/python3.5/dist-packages/airflow/models/init.py”，第 1441 行结果 = task_copy.execute(context=context) 文件“/usr/local/lib/python3.5/dist-packages/airflow/operators/papermill_operator.py”，第 63 行，在执行参数=self.inlets[i].parameters, getattr 中的文件“/usr/local/lib/python3.5/dist-packages/airflow/lineage/datasets.py”，第 66 行 return env.from_string(self._data.get(attr)).render(**self.context) 文件“/home/exa00112/.local/lib/python3.5/site-packages/jinja2/environment.py”，第 880 行，在 from_string return cls.from_code(self, self.compile(source), globals, None) 编译中的文件“/home/exa00112/.local/lib/python3.5/site-packages/jinja2/environment.py”，第 581 行 defer_init=defer_init) _generate 中的文件“/home/exa00112/.local/lib/python3.5/site-packages/jinja2/environment.py”，第 543 行优化=自我优化）文件“/home/exa00112/.local/lib/python3.5/site-packages/jinja2/compiler.py”，第 78 行，在生成中 raise TypeError('不能编译非模板节点') TypeError：无法编译非模板节点 [2019-09-10 20:36:48,808] {init.py:1611} INFO - 将任务标记为 FAILED。

【问题讨论】：

标签： jupyter-notebook papermill airflow

【解决方案1】：

似乎是 Papermill 操作员向 Papermill 传递无效参数数据结构（在 Papermill 查找 dict 时传递字符串参数 dict）到

的已知问题

https://issues.apache.org/jira/browse/AIRFLOW-5774

不确定什么时候会修复，因为它看起来是重复的问题，很难追踪

【讨论】：