连接到 Lakebase

使用结构化流写入 Lakebase 或带有内置批处理、自动重试和工作区管理认证的外部 PostgreSQL 数据库。

何时使用 Lakebase 接收端

使用 Lakebase 接收器，将数据以低延迟流式写入 Lakebase 或外部 PostgreSQL 数据库。此接收端无需你实现自定义 foreach 函数来处理批处理、连接管理和错误处理。

常见用例包括：

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

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

要求

Databricks 运行时 18 LTS 及以上版本。
- 外部 PostgreSQL 连接要求使用 Databricks Runtime 19 及更高版本，并加入 Custom JDBC on UC Compute 预览。
- 区间数据类型需要使用 Databricks Runtime 19 及以上版本。
采用专用或标准访问模式的经典型计算，或适用于笔记本或作业的无服务器计算。在无服务器计算中，使用 Trigger.AvailableNow()。请参阅无服务器计算中的流式处理。
Lakebase 数据库，或通过 Unity 目录连接到外部 PostgreSQL 数据库。

标识符要求

对于所有目标，Databricks建议使用模式、表、列和主键列名，这些名称以字母或下划线开头，且仅包含字母、数字和下划线。当汇总库自动创建Lakebase表时，会强制执行这些要求。要使用不满足这些要求的标识符，请在开始查询前创建目标表。

连接到数据库

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

已在 Unity Catalog 中注册的 Lakebase 表

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

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

要写入 Lakebase 表，请使用 .toTable() 方法，并提供完全限定表名：catalog.schema.table

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-columns>")  // Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

替换以下占位符：

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

关于可选配置，如 batchsize 和 batchinterval，请参见 PostgreSQL的汇入选项。

未在 Unity Catalog 中注册的 Lakebase 表

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

要写入 Lakebase 表，请使用 endpoint 和 dbtable 选项：

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-columns>")  # Optional
  .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-columns>")  // Optional
  .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>：可选。目标PostgreSQL数据库的名称。默认值为 databricks_postgres. 请参阅 “管理数据库”。
<schema>.<table>：schema.table 格式的目标表。如果省略架构，接收器将使用 public 架构。自动创建表格时，使用以字母或下划线开头的标识符，仅包含字母、数字和下划线。
<primary-key-columns>：可选。目标表的主键中所有列的逗号分隔列表，例如 id 或 user_id,event_type。参见 Upsert 行为。
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>：查询用于存储其检查点的 Unity Catalog 卷路径。还可以使用云对象存储 URI。该位置必须是可以写入到的存储，而不是本地磁盘，并且每个流式处理查询必须是唯一的。这与目标表无关。请参阅结构化流式处理检查点。

关于可选配置，如 batchsize 和 batchinterval，请参见 PostgreSQL的汇入选项。

带有 Unity 目录凭证的外部 PostgreSQL

Important

此功能目前以公共预览版提供。工作区管理员可以通过预览页面控制 UC Compute 上自定义 JDBC 的访问权限。请参阅 Manage Azure Databricks 预览版。

使用Unity Catalog连接，在代码中不存储凭证的情况下，向外部PostgreSQL数据库进行认证。目标表必须已经存在。

创建类型的 POSTGRESQL连接，参见创建连接。执行查询的用户或服务主体必须拥有对该连接的 USE CONNECTION。

要写入PostgreSQL表，请使用 databricks.connection、 database和 dbtable 选项：

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  // Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

替换以下占位符：

<connection-name>：Unity 目录连接的名称。
<database>：目标 PostgreSQL 数据库的名称。
<schema>.<table>：采用 schema.table 格式的现有目标表。如果省略架构，接收器将使用 public 架构。
<primary-key-columns>：可选。目标表的主键中所有列的逗号分隔列表，例如 id 或 user_id,event_type。参见 Upsert 行为。
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>：查询用于存储其检查点的 Unity Catalog 卷路径。还可以使用云对象存储 URI。该位置必须是可以写入到的存储，而不是本地磁盘，并且每个流式处理查询必须是唯一的。这与目标表无关。请参阅结构化流式处理检查点。

PostgreSQL连接总是使用TLS。证书验证遵循你创建连接时选择的Unity Catalog连接设置：

信任服务器证书：选择时，连接使用 sslmode=require，该连接在不验证服务器证书的情况下加密连接。
用户提供的服务器证书：提供一个PEM编码的服务器证书，以便在未选择信任服务器证书时使用sslmode=verify-full。如果你不提供证书，该连接将使用 sslmode=verify-full 以及 JVM 默认信任存储。

配置选项

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

关于汇槽配置选项，包括通用选项和每种连接方法的选项，请参见 PostgreSQL汇槽选项。

数据类型映射

汇入器在写入现有 Lakebase 或外部 PostgreSQL 表前，检查每个 DataFrame 列是否与其对应的目标列兼容。

下表包含了Databricks Runtime 18 LTS及以上版本支持的类型：

Spark 类型	自动创建的 Lakebase 表类型	现有PostgreSQL表中的兼容类型
`ByteType`、`ShortType`	`smallint`	`smallint`
`IntegerType`	`integer`	`integer`
`LongType`	`bigint`	`bigint`
`FloatType`	`real`	`real`
`DoubleType`	`double precision`	`double precision`
`DecimalType`	`numeric`	`numeric`
`StringType`	`text`	`varchar`、`text`
`VarcharType(n)`	`varchar(n)`	`varchar`、`text`
`CharType(n)`	`char(n)`	`char`
`BinaryType`	`bytea`	`bytea`
`BooleanType`	`boolean`	`boolean`
`TimestampType`	`timestamptz`	`timestamptz`
`TimestampNTZType`	`timestamp`	`timestamp`
`DateType`	`date`	`date`
`ArrayType`、`MapType`、`StructType`、`VariantType`、`NullType`	`jsonb`	`json`、`jsonb`

下表包含了Databricks Runtime 19及以上版本支持的类型：

Spark 类型	自动创建的 Lakebase 表类型	现有PostgreSQL表中的兼容类型
`DayTimeIntervalType`、`YearMonthIntervalType`	`interval`	`interval`

Upsert 行为

该 upsertkey 选项用于识别目标表的主键列。对于已有的表，列 upsertkey 必须与表的主键完全匹配。如果省略该选项，接收器会从表中读取主键。对于由 sink 创建的 Lakebase 表，upsertkey 定义了主键。如果你省略了这个选项，水槽会生成没有主键的表格。

当目标表具有主键时，sink 会使用 PostgreSQL 的 INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... 语法执行 upsert。当目标表没有主键时，汇入器执行插入。查询的输出模式对此行为没有影响。

所有主键列必须存在于数据帧中，并使用可比较的类型，如数字类型或字符串类型。

性能调优

批处理和回压

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

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

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

延迟和吞吐量指南：

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

连接行为

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

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

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

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

支持的触发器和输出模式

Triggers

下表显示了在经典和无服务器计算上对结构化流触发类型的支持：

触发器	经典计算	无服务器计算（笔记本和作业）
`RealTime`	Yes	否
`ProcessingTime`	Yes	否
`AvailableNow`	Yes	Yes
`Once`	Yes. Deprecated. 使用 `AvailableNow`。	Yes. Deprecated. 使用 `AvailableNow`。

输出模式

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

输出模式	支持
`update`	Yes
`append`	Yes. 其行为与 `update` 相同。当目标表具有主键时，查询将执行更新插入；否则，查询将执行插入。参见 Upsert 行为。
`complete`	否

限制

对于通过 Unity 目录连接的外部 PostgreSQL 数据库，目标表必须已经存在。 Sink 只在 Lakebase 中自动创建缺失的表。
不支持湖流管道。

反馈

此页面是否有帮助？

上次更新时间为 2026-10-01

连接到 Lakebase

何时使用 Lakebase 接收端

要求

标识符要求

连接到数据库

已在 Unity Catalog 中注册的 Lakebase 表

Python

Scala

未在 Unity Catalog 中注册的 Lakebase 表

Python

Scala

带有 Unity 目录凭证的外部 PostgreSQL

Python

Scala

配置选项

数据类型映射

Upsert 行为

性能调优

批处理和回压

连接行为

支持的触发器和输出模式

Triggers

输出模式

限制

反馈

其他资源