使用 YDBExecuteQueryOperator 操作 YDB 的操作指南¶

简介¶

Apache Airflow 拥有丰富的操作符库，可用于实现构成工作流的各种任务。Airflow 本质上是一个由任务（节点）和依赖关系（边）组成的图（有向无环图）。

由操作符定义或实现的任务是数据管道中的一个工作单元。

本指南的目的是使用 YDBExecuteQueryOperator 定义涉及与 YDB 数据库交互的任务。

使用 YDBExecuteQueryOperator 进行常见的数据库操作¶

YDBExecuteQueryOperator 执行 DML 或 DDL 查询。操作符的参数为

sql - 包含查询的字符串；
is_ddl - 指示查询是否为 DDL 的标志。默认为 false；
conn_id - YDB 连接 ID。默认值为 ydb_default；
params - 如果查询是 Jinja 模板，则要注入到查询中的参数，有关 params 的更多详细信息

注意

参数 is_ddl 可能会在操作符的未来版本中删除。

创建 YDB 表¶

以下代码片段基于 Airflow-2.0

tests/system/providers/ydb/example_ydb.py[源代码]

# create_pet_table, populate_pet_table, get_all_pets, and get_birth_date are examples of tasks created by
# instantiating the YDB Operator

ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
DAG_ID = "ydb_operator_dag"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime.datetime(2020, 2, 2),
    schedule="@once",
    catchup=False,
) as dag:
    create_pet_table = YDBExecuteQueryOperator(
        task_id="create_pet_table",
        sql="""
            CREATE TABLE pet (
            pet_id INT,
            name TEXT NOT NULL,
            pet_type TEXT NOT NULL,
            birth_date TEXT NOT NULL,
            owner TEXT NOT NULL,
            PRIMARY KEY (pet_id)
            );
          """,
        is_ddl=True,  # must be specified for DDL queries
    )

将 SQL 语句直接放入操作符中并不十分吸引人，并且会在以后的某个时候造成可维护性问题。为了防止这种情况，Airflow 提供了一种优雅的解决方案。其工作原理如下：您只需在 DAG 文件夹中创建一个名为 sql 的目录，然后将包含 SQL 查询的所有 SQL 文件放入其中。

您的 dags/sql/pet_schema.sql 应该如下所示

-- create pet table
CREATE TABLE pet (
  pet_id INT,
  name TEXT NOT NULL,
  pet_type TEXT NOT NULL,
  birth_date TEXT NOT NULL,
  owner TEXT NOT NULL,
  PRIMARY KEY (pet_id)
);

现在让我们重构 DAG 中的 create_pet_table

create_pet_table = YDBExecuteQueryOperator(
    task_id="create_pet_table",
    sql="sql/pet_schema.sql",
)

将数据插入 YDB 表¶

假设我们在 dags/sql/pet_schema.sql 文件中已经有了以下 SQL 插入语句

-- populate pet table
UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
VALUES (1, 'Max', 'Dog', '2018-07-05', 'Jane');

UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
VALUES (2, 'Susie', 'Cat', '2019-05-01', 'Phil');

UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
VALUES (3, 'Lester', 'Hamster', '2020-06-23', 'Lily');

UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
VALUES (4, 'Quincy', 'Parrot', '2013-08-11', 'Anne');

然后，我们可以创建一个填充 pet 表的 YDBExecuteQueryOperator 任务。

populate_pet_table = YDBExecuteQueryOperator(
    task_id="populate_pet_table",
    sql="sql/pet_schema.sql",
)

从 YDB 表中获取记录¶

从 YDB 表中获取记录可以像下面这样简单

get_all_pets = YDBExecuteQueryOperator(
    task_id="get_all_pets",
    sql="SELECT * FROM pet;",
)

将参数传递给 YDBExecuteQueryOperator¶

BaseOperator 类具有 params 属性，YDBExecuteQueryOperator 可以通过继承获得该属性。params 可以通过许多有趣的方式动态传递参数。

要查找名为“Lester”的宠物的主人

get_birth_date = YDBExecuteQueryOperator(
    task_id="get_birth_date",
    sql="SELECT * FROM pet WHERE birth_date BETWEEN '{{params.begin_date}}' AND '{{params.end_date}}'",
    params={"begin_date": "2020-01-01", "end_date": "2020-12-31"},
)

现在让我们重构 get_birth_date 任务。我们不要将 SQL 语句直接放入代码中，而是通过创建一个 sql 文件来整理一下。

-- dags/sql/birth_date.sql
SELECT * FROM pet WHERE birth_date BETWEEN '{{params.begin_date}}' AND '{{params.end_date}}';

get_birth_date = YDBExecuteQueryOperator(
    task_id="get_birth_date",
    sql="sql/birth_date.sql",
    params={"begin_date": "2020-01-01", "end_date": "2020-12-31"},
)

完整的 YDB 操作符 DAG¶

当我们把所有东西放在一起时，我们的 DAG 应该是这样的

tests/system/providers/ydb/example_ydb.py[源代码]

# create_pet_table, populate_pet_table, get_all_pets, and get_birth_date are examples of tasks created by
# instantiating the YDB Operator

ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
DAG_ID = "ydb_operator_dag"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime.datetime(2020, 2, 2),
    schedule="@once",
    catchup=False,
) as dag:
    create_pet_table = YDBExecuteQueryOperator(
        task_id="create_pet_table",
        sql="""
            CREATE TABLE pet (
            pet_id INT,
            name TEXT NOT NULL,
            pet_type TEXT NOT NULL,
            birth_date TEXT NOT NULL,
            owner TEXT NOT NULL,
            PRIMARY KEY (pet_id)
            );
          """,
        is_ddl=True,  # must be specified for DDL queries
    )
    populate_pet_table = YDBExecuteQueryOperator(
        task_id="populate_pet_table",
        sql="""
              UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
              VALUES (1, 'Max', 'Dog', '2018-07-05', 'Jane');

              UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
              VALUES (2, 'Susie', 'Cat', '2019-05-01', 'Phil');

              UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
              VALUES (3, 'Lester', 'Hamster', '2020-06-23', 'Lily');

              UPSERT INTO pet (pet_id, name, pet_type, birth_date, owner)
              VALUES (4, 'Quincy', 'Parrot', '2013-08-11', 'Anne');
            """,
    )
    get_all_pets = YDBExecuteQueryOperator(task_id="get_all_pets", sql="SELECT * FROM pet;")
    get_birth_date = YDBExecuteQueryOperator(
        task_id="get_birth_date",
        sql="SELECT * FROM pet WHERE birth_date BETWEEN '{{params.begin_date}}' AND '{{params.end_date}}'",
        params={"begin_date": "2020-01-01", "end_date": "2020-12-31"},
    )

    create_pet_table >> populate_pet_table >> get_all_pets >> get_birth_date

结论¶

在本操作指南中，我们探讨了 Apache Airflow YDBExecuteQueryOperator 如何连接到 YDB 数据库。让我们快速回顾一下要点。最佳实践是在 dags 目录中创建一个名为 sql 的子目录，您可以在其中存储 sql 文件。这将使您的代码更优雅、更易于维护。最后，我们查看了 sql 脚本的模板化版本以及 params 属性的用法。