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 屬性的使用。