连接到 Lakebase

使用 Structured Streaming 向 Lakebase 写入数据，该功能具有内置批处理、自动重试以及由工作区管理的身份验证。

何时使用 Lakebase 接收端

使用 Lakebase 接收器向 Lakebase 进行低延迟流式写入。 此接收端无需你实现自定义 foreachBatch 函数来处理批处理、连接管理和错误处理。

常见用例包括：

• 为运营仪表板或面向客户的功能实时更新应用程序数据库。
• 将持续更改的数据（例如聚合或筛选的流式处理结果）同步到事务数据库。
• 使用 实时模式以亚秒级延迟将 Structured Streaming 查询的输出写入 Lakebase 表。

若要以相反方向将数据从 Lakebase 同步到 Lakehouse 中的 Delta Lake 表，请参阅 Lakebase 变更数据馈送。

要求

• Databricks Runtime 18 及更高版本
• 具有专用或标准访问模式的经典计算。
• Lakebase 数据库

连接到数据库

Lakebase 接收器支持以下连接方法：

已在 Unity Catalog 中注册的 Lakebase 表

对于注册到 Unity 目录的 Lakebase 表，连接器会自动管理凭据，并使用运行查询的用户或服务主体的标识。 如果该表不存在，连接器将创建该表。

若要向 Unity 目录注册 Lakebase 数据库，请参阅 在 Unity 目录中注册 Lakebase 数据库。

若要向 Lakebase 表中写入数据，请使用 .toTable() 方法，并指定完全限定表名 catalog.schema.table。 以下示例显示了所需的选项以及可选 upsertkey 选项：

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

替换以下占位符：

• <catalog>.<schema>.<table>：目标表的完全限定名称。 catalog这是在注册 Lakebase 数据库时创建的 Unity 目录，请参阅在 Unity 目录中注册 Lakebase 数据库。 如果该表不存在，连接器会创建该表。
• <primary-key-column>：可选。 以逗号分隔的构成 upsert 键的列列表，例如 id 或 user_id,event_type。 如果省略 upsertkey，接收器将从目标表的主键推断密钥。 请参阅 Upsert 的行为。
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>：查询用于存储其检查点的 Unity Catalog 卷路径。 还可以使用云对象存储 URI。 该位置必须是可以写入到的存储，而不是本地磁盘，并且每个流式处理查询必须是唯一的。 这与目标表无关。 请参阅结构化流式处理检查点。

有关可选配置，例如 batchsize 和 batchinterval，请参阅 配置选项。

未在 Unity Catalog 中注册的 Lakebase 表

对于未注册到 Unity 目录的 Lakebase 表，连接器会自动管理凭据，并使用运行查询的用户或服务主体的标识。 如果该表不存在，连接器将创建该表。

若要写入 Lakebase 表，请使用 endpoint 和 dbtable 选项。 以下示例还包括可选的 database 和 upsertkey 选项：

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  # Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  // Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

替换以下占位符：

• <project-id>.<branch-id>.<endpoint-id>：您的 Lakebase 端点。 在“计算”选项卡的“获取 ID”菜单上查找资源名称中的所有三个值，其格式projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>为 。 请参阅 计算标识符。
• <database>：可选。 目标 Postgres 数据库的名称。 默认值为 databricks_postgres. 请参阅 “管理数据库”。
• <schema>.<table>：schema.table 格式的目标表。 如果省略架构，接收器将使用 public 架构。 使用以字母或下划线开头且仅包含字母、数字和下划线的简单标识符;不支持带引号的标识符和特殊字符（如连字符）。
• <primary-key-column>：可选。 以逗号分隔的构成 upsert 键的列列表，例如 id 或 user_id,event_type。 如果省略 upsertkey，接收器将从目标表的主键推断密钥。 请参阅 Upsert 的行为。
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>：查询用于存储其检查点的 Unity Catalog 卷路径。 还可以使用云对象存储 URI。 该位置必须是可以写入到的存储，而不是本地磁盘，并且每个流式处理查询必须是唯一的。 这与目标表无关。 请参阅结构化流式处理检查点。

有关可选配置，例如 batchsize 和 batchinterval，请参阅 配置选项。

配置选项

对于无法识别的选项，接收器会报告错误 JDBC_STREAMING_SINK_INVALID_OPTIONS。

以下选项适用于所有连接方法：

未在 Unity Catalog 中注册的 Lakebase 表

当您连接到未在 Unity Catalog 中注册的 Lakebase 表时，以下选项适用：

Upsert 行为

当存在更新插入键（无论是通过 upsertkey 指定，还是由接收器根据表的主键推断得出）时，接收器将使用 PostgreSQL 的 INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... 语法对表执行更新插入操作。

当不存在更新插入键时，接收器执行插入操作。 查询的输出模式不会影响更新插入或插入行为。

upsertkey 列必须：

• 是 DataFrame 列的非空子集。
• 与目标表的 PRIMARY KEY 完全匹配。 如果指定的列与主键不匹配，查询将失败。
• 是可比较的类型，例如数值类型或字符串类型。 为防止并发写入期间发生数据库死锁，接收端会在每个批次内按更新插入键对行进行排序。 Upsert 键不支持复杂类型或结构类型。

列名称会自动使用 PostgreSQL 默认的双引号 " 进行引用，以处理保留关键字和大小写混合的名称。

表和架构名称必须使用以字母或下划线开头且仅包含字母、数字和下划线的简单标识符。 接收端不支持表名或模式名中包含带引号的标识符或特殊字符（如连字符）。

性能优化

批处理和回压

满足以下任一条件时将触发刷新：

• 缓冲区达到 batchsize 行，其默认值为 1000。
• 缓冲区期限超过 batchinterval，默认为 100 milliseconds。

当数据库无法跟上传入数据速率时，接收端会将反压向上游传播至源端。

延迟和吞吐量指南：

• 对于采用实时模式的低延迟工作负荷，请将 batchinterval 值调低，以确保更短的最大刷新间隔时间。 有关代码示例，请参阅 结构化流式处理中的实时模式 ，了解概念和 实时模式示例 。
• 对于高吞吐量工作负荷，增加 batchsize 以减少每个事务的开销。

连接行为

该接收器在执行器上使用连接池。 默认情况下，每个任务使用一个数据库连接。

Databricks 建议对每个连接使用任务默认值 1。 如果增加每个连接的任务数，可能会导致连接争用，并增加高吞吐量连接的延迟。

若要配置任务与连接的比率，请设置 spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark 配置。 如果目标数据库的连接限制较低，请减少随机分区数或增加 spark.databricks.sql.streaming.jdbc.tasksPerConnection。

该接收器会自动重试暂时的 JDBC 错误，包括连接失败、死锁和速率限制。 如果接收端用尽所有重试次数，查询将失败。

支持的触发器和输出模式

Triggers

下表显示了对结构化流处理触发器类型的支持情况：

输出模式

下表显示了对结构化流式处理输出模式的支持：

限制

• 不支持无服务器计算和 Lakeflow 管道。
• 仅支持 Lakebase 作为写入目标。 不支持与外部 PostgreSQL 兼容的数据库。