将转换后的oozie DAG部署到Google Composer中：没有名为“o2a”的模块

Question

我正在使用google oozie to airflow converter转换一些在AWS EMR上运行的oozie工作流。设法获得了第一个版本，但当我尝试上载DAG时，airflow抛出了一个错误：

损坏的DAG:没有名为“o2a”的模块

我已经尝试部署pypi包o2a，两者都使用命令

gcloud composer environments update composer-name --update-pypi-packages-from-file requirements.txt --location location

从谷歌云控制台。两者都失败了

requirements.txt

o2a==1.0.1

这是密码

    from airflow import models
    from airflow.operators.subdag_operator import SubDagOperator
    from airflow.utils import dates
    from o2a.o2a_libs import functions
    from airflow.models import Variable

    import subdag_validation
    import subdag_generate_reports

    CONFIG = {}

    JOB_PROPS = {

    }

    dag_config = Variable.get("coordinator", deserialize_json=True)
    cdrPeriod = dag_config["cdrPeriod"]

    TASK_MAP = {"validation": ["validation"], "generate_reports": ["generate_reports"] }

    TEMPLATE_ENV = {**CONFIG, **JOB_PROPS, "functions": functions, "task_map": TASK_MAP}

    with models.DAG(
        "workflow_coordinator",
        schedule_interval=None,  # Change to suit your needs
        start_date=dates.days_ago(0),  # Change to suit your needs
        user_defined_macros=TEMPLATE_ENV,
    ) as dag:

        validation = SubDagOperator(
            task_id="validation",
            trigger_rule="one_success",
            subdag=subdag_validation.sub_dag(dag.dag_id, "validation", dag.start_date, dag.schedule_interval),
        )

        generate_reports = SubDagOperator(
            task_id="generate_reports",
            trigger_rule="one_success",
            subdag=subdag_generate_reports.sub_dag(dag.dag_id, "generate_reports", dag.start_date, dag.schedule_interval,
            {
                "cdrPeriod": "{{cdrPeriod}}"
            }),
        )

        validation.set_downstream(generate_reports)

Answer 1

o2a文档中有一节介绍了如何部署o2a：

https://github.com/GoogleCloudPlatform/oozie-to-airflow#the-o2a-libraries

由于另一个依赖项：lark parser，启动失败 刚刚使用pypi软件包管理器安装的Composer实现了这一点