YDB Operator 使用指南¶

介绍¶

Apache Airflow 拥有一套强大的 Operator 库，可用于实现组成工作流程的各种任务。Airflow 本质上是一个由任务（节点）和依赖（边）组成的图（有向无环图）。

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

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

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

YDBExecuteQueryOperator 执行 DML 或 DDL 查询。该 Operator 的参数如下：

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

注意

参数 is_ddl 可能在 Operator 的未来版本中移除。

创建 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)


def sanitize_date(value: str) -> str:
    """Ensure the value is a valid date format"""
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        raise ValueError(f"Invalid date format: {value}")
    return value


def transform_dates(**kwargs):
    begin_date = sanitize_date(kwargs.get("begin_date"))
    end_date = sanitize_date(kwargs.get("end_date"))
    return {"begin_date": begin_date, "end_date": end_date}


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 语句直接嵌入 Operator 中不够吸引人，并且会在未来导致维护上的困难。为避免此问题，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 Operator 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)


def sanitize_date(value: str) -> str:
    """Ensure the value is a valid date format"""
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        raise ValueError(f"Invalid date format: {value}")
    return value


def transform_dates(**kwargs):
    begin_date = sanitize_date(kwargs.get("begin_date"))
    end_date = sanitize_date(kwargs.get("end_date"))
    return {"begin_date": begin_date, "end_date": end_date}


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;")
    transform_dates = PythonOperator(
        task_id="transform_dates",
        python_callable=transform_dates,
        op_kwargs={"begin_date": "{{params.begin_date}}", "end_date": "{{params.end_date}}"},
        params={"begin_date": "2020-01-01", "end_date": "2020-12-31"},
    )
    get_birth_date = YDBExecuteQueryOperator(
        task_id="get_birth_date",
        sql="""
        SELECT * FROM pet WHERE birth_date BETWEEN '{{ ti.xcom_pull(task_ids="transform_dates")["begin_date"] }}' AND '{{ ti.xcom_pull(task_ids="transform_dates")["end_date"] }}'
        """,
    )

    (
        create_pet_table
        >> populate_pet_table
        >> populate_pet_table_via_bulk_upsert()
        >> get_all_pets
        >> transform_dates
        >> get_birth_date
    )

结论¶

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