使用 YDB 运算符的操作指南¶

简介¶

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

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

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

YDBExecuteQueryOperator 的常见数据库操作¶

YDBExecuteQueryOperator 执行 DML 或 DDL 查询。运算符的参数为

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

注意

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

创建 YDB 表¶

以下代码片段基于 Airflow-2.0

tests/system/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"


@task
def populate_pet_table_via_bulk_upsert():
    hook = YDBHook()
    column_types = (
        ydb.BulkUpsertColumns()
        .add_column("pet_id", ydb.OptionalType(ydb.PrimitiveType.Int32))
        .add_column("name", ydb.PrimitiveType.Utf8)
        .add_column("pet_type", ydb.PrimitiveType.Utf8)
        .add_column("birth_date", ydb.PrimitiveType.Utf8)
        .add_column("owner", ydb.PrimitiveType.Utf8)
    )

    rows = [
        {"pet_id": 3, "name": "Lester", "pet_type": "Hamster", "birth_date": "2020-06-23", "owner": "Lily"},
        {"pet_id": 4, "name": "Quincy", "pet_type": "Parrot", "birth_date": "2013-08-11", "owner": "Anne"},
    ]
    hook.bulk_upsert("pet", rows=rows, column_types=column_types)


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');

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

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/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"


@task
def populate_pet_table_via_bulk_upsert():
    hook = YDBHook()
    column_types = (
        ydb.BulkUpsertColumns()
        .add_column("pet_id", ydb.OptionalType(ydb.PrimitiveType.Int32))
        .add_column("name", ydb.PrimitiveType.Utf8)
        .add_column("pet_type", ydb.PrimitiveType.Utf8)
        .add_column("birth_date", ydb.PrimitiveType.Utf8)
        .add_column("owner", ydb.PrimitiveType.Utf8)
    )

    rows = [
        {"pet_id": 3, "name": "Lester", "pet_type": "Hamster", "birth_date": "2020-06-23", "owner": "Lily"},
        {"pet_id": 4, "name": "Quincy", "pet_type": "Parrot", "birth_date": "2013-08-11", "owner": "Anne"},
    ]
    hook.bulk_upsert("pet", rows=rows, column_types=column_types)


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');
            """,
    )
    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
        >> populate_pet_table_via_bulk_upsert()
        >> get_all_pets
        >> get_birth_date
    )

结论¶

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