DatabricksSqlOperator¶

使用 DatabricksSqlOperator 在 Databricks SQL warehouse 或 Databricks 集群上执行 SQL。

使用操作符¶

该操作符针对配置的 warehouse 执行给定的 SQL 查询。唯一必需的参数是

sql - 要执行的 SQL 查询。有 3 种指定 SQL 查询的方式
1. 包含 SQL 语句的简单字符串。
2. 表示 SQL 语句的字符串列表。
3. 包含 SQL 查询的文件名。文件必须具有 .sql 扩展名。每个查询应以 ;<new_line> 结尾
sql_warehouse_name (要使用的 Databricks SQL warehouse 名称) 或 http_path (Databricks SQL warehouse 或 Databricks cluster 的 HTTP 路径) 之一。

其他参数是可选的，可以在类文档中找到。

示例¶

选择数据¶

使用 DatabricksSqlOperator 从表中选择数据的示例用法如下

tests/system/databricks/example_databricks_sql.py

    # Example of using the Databricks SQL Operator to select data.
    select = DatabricksSqlOperator(
        databricks_conn_id=connection_id,
        sql_endpoint_name=sql_endpoint_name,
        task_id="select_data",
        sql="select * from default.my_airflow_table",
    )

将数据选择到文件¶

使用 DatabricksSqlOperator 从表中选择数据并存储到文件的示例用法如下

tests/system/databricks/example_databricks_sql.py

    # Example of using the Databricks SQL Operator to select data into a file with JSONL format.
    select_into_file = DatabricksSqlOperator(
        databricks_conn_id=connection_id,
        sql_endpoint_name=sql_endpoint_name,
        task_id="select_data_into_file",
        sql="select * from default.my_airflow_table",
        output_path="/tmp/1.jsonl",
        output_format="jsonl",
    )

执行多个语句¶

使用 DatabricksSqlOperator 执行多个 SQL 语句的示例用法如下

tests/system/databricks/example_databricks_sql.py

    # Example of using the Databricks SQL Operator to perform multiple operations.
    create = DatabricksSqlOperator(
        databricks_conn_id=connection_id,
        sql_endpoint_name=sql_endpoint_name,
        task_id="create_and_populate_table",
        sql=[
            "drop table if exists default.my_airflow_table",
            "create table default.my_airflow_table(id int, v string)",
            "insert into default.my_airflow_table values (1, 'test 1'), (2, 'test 2')",
        ],
    )

从文件执行多个语句¶

使用 DatabricksSqlOperator 执行文件中的语句的示例用法如下

tests/system/databricks/example_databricks_sql.py

    # Example of using the Databricks SQL Operator to select data.
    # SQL statements should be in the file with name test.sql
    create_file = DatabricksSqlOperator(
        databricks_conn_id=connection_id,
        sql_endpoint_name=sql_endpoint_name,
        task_id="create_and_populate_from_file",
        sql="test.sql",
    )

DatabricksSqlSensor¶

使用 DatabricksSqlSensor 运行传感器，以通过 Databricks SQL warehouse 或交互式集群访问表。

使用传感器¶

传感器执行用户提供的 SQL 语句。唯一必需的参数是

sql - 传感器要执行的 SQL 查询。
sql_warehouse_name (要使用的 Databricks SQL warehouse 名称) 或 http_path (Databricks SQL warehouse 或 Databricks cluster 的 HTTP 路径) 之一。

其他参数是可选的，可以在类文档中找到。

示例¶

配置要与传感器一起使用的 Databricks 连接。

tests/system/databricks/example_databricks_sensors.py

# Connection string setup for Databricks workspace.
connection_id = "databricks_default"
sql_warehouse_name = "Starter Warehouse"

使用 SQL 语句探测指定的表

tests/system/databricks/example_databricks_sensors.py

# Example of using the Databricks SQL Sensor to check existence of data in a table.
sql_sensor = DatabricksSqlSensor(
    databricks_conn_id=connection_id,
    sql_warehouse_name=sql_warehouse_name,
    catalog="hive_metastore",
    task_id="sql_sensor_task",
    sql="select * from hive_metastore.temp.sample_table_3 limit 1",
    timeout=60 * 2,
)

DatabricksPartitionSensor¶

传感器是一种特殊类型的操作符 (Operator)，其设计目标是只做一件事 - 等待某事发生。它可以是基于时间的等待，也可以是等待文件或外部事件，但它们所做的就是等待直到某事发生，然后成功，以便它们的下游任务可以运行。

对于 Databricks Partition Sensor，我们检查分区及其相关值是否存在，如果不存在，它会等待直到分区值到达。等待时间和检查间隔可以分别在 timeout 和 poke_interval 参数中配置。

使用 DatabricksPartitionSensor 运行传感器，以通过 Databricks SQL warehouse 或交互式集群访问表。

使用传感器¶

传感器接受用户提供的表名和分区名称/值，并生成 SQL 查询来检查指定的分区名称/值是否存在于指定的表中。

必需的参数是

table_name (用于检查分区的表的名称)。
partitions (要检查的分区的名称)。
partition_operator (分区的比较操作符，用于值的范围或限制，例如 partition_name >= partition_value)。支持 Databricks 比较操作符。
sql_warehouse_name (要使用的 Databricks SQL warehouse 名称) 或 http_path (Databricks SQL warehouse 或 Databricks cluster 的 HTTP 路径) 之一。

其他参数是可选的，可以在类文档中找到。

示例¶

配置要与传感器一起使用的 Databricks 连接。

tests/system/databricks/example_databricks_sensors.py

# Connection string setup for Databricks workspace.
connection_id = "databricks_default"
sql_warehouse_name = "Starter Warehouse"

探测特定表是否存在数据/分区

tests/system/databricks/example_databricks_sensors.py

# Example of using the Databricks Partition Sensor to check the presence
# of the specified partition(s) in a table.
partition_sensor = DatabricksPartitionSensor(
    databricks_conn_id=connection_id,
    sql_warehouse_name=sql_warehouse_name,
    catalog="hive_metastore",
    task_id="partition_sensor_task",
    table_name="sample_table_2",
    schema="temp",
    partitions={"date": "2023-01-03", "name": ["abc", "def"]},
    partition_operator="=",
    timeout=60 * 2,
)