气流：如何将读取 json 文件的方法放在本地库中

Question

我必须生成一些 dag。我已将 json 表模式文件保存在 GCP bucket 上。 与 composer 关联的 GCP 存储桶上的文件将重新映射到 /home/airflow/gcs/dags/。 如果我定义读取 json 文件的方法，在创建 dag 之后，一切正常。 但是如果我希望生成一些“通用代码”（放在我的库中），我无法使用库中的代码访问 FileSystem，具体我无法使用 python json 库。

奇怪的是，我在dag创建步骤之外定义了方法，但我只是在dag创建之后才调用它！

为了完成讨论，如果库中的代码仅在内存对象中使用，我没有问题。

我在使用气流时遇到了这个问题（composer 的 GCP 驱动程序为 1.9）

这是我的外部库：

lib/
    __init__.py
    bb_airflow_utils.py

关于外部库

def load_json_file(fname):
    #per far sì che il dag la veda
    with open(fname, 'r') as f:
        d = json.load(f)
    return d

关于主脚本

from lib.bb_airflow_utils import *
ROOT_PATH = 'home/airflow/gcs/dags'
IDCLI = 'goofy'
...
...
with DAG(dag_id=dag_name, default_args=dag_args) as dag:
    filepath = path.join(ROOT_PATH, '{}-todwh.json'.format(IDCLI))
    get_data = load_json_file(filepath)
    .....
    task_dummy_start = DummyOperator(task_id='task_{}_start'.format(dag_name), dag=dag)
    .....

Airflow 无视运营商并通过 UI 表示 dag 还没有 SLA

Answer 1

看看https://cloud.google.com/composer/docs/how-to/using/installing-python-dependencies#install-local。

您可以将通用代码放在一个单独的文件中，然后将其放在单独的文件夹中，如下例所示。

将依赖项放在dags/ 文件夹的子目录中。要从子目录导入模块，模块路径中的每个子目录都必须包含一个__init__.py 包标记文件。

在这个例子中，依赖是 coin_module.py：

dags/
  use_local_deps.py  # A DAG file.
  dependencies/
    __init__.py
    coin_module.py

从 DAG 定义文件中导入依赖项。

例如：

from dependencies import coin_module