Connect to Lakebase

Use Structured Streaming to write to Lakebase or an external PostgreSQL database with built-in batching, automatic retries, and workspace-managed authentication.

When to use the Lakebase sink

Use the Lakebase sink for low-latency streaming writes to Lakebase or an external PostgreSQL database. This sink doesn't require you to implement custom foreach functions to handle batching, connection management, and error handling.

Common use cases include:

  • Update application databases in real-time for operational dashboards or customer-facing features.
  • Sync continuously changing data, such as aggregated or filtered streaming results, into a transactional database.
  • Write the output of a Structured Streaming query into a Lakebase table with sub-second latency using real-time mode.

To sync data from Lakebase to Delta Lake tables in the Lakehouse, the reverse direction, see Lakebase Change Data Feed.

Requirements

  • Databricks Runtime 18 LTS and above.
  • Classic compute with dedicated or standard access modes, or serverless compute for notebooks or jobs. On serverless compute, use Trigger.AvailableNow(). See Streaming on serverless compute.
  • A Lakebase database, or a Unity Catalog connection to an external PostgreSQL database.

Identifier requirements

For all targets, Databricks recommends using schema, table, column, and primary-key column names that start with a letter or underscore and contain only letters, numbers, and underscores. The sink enforces these requirements when it automatically creates a Lakebase table. To use identifiers that don't meet these requirements, create the target table before starting the query.

Connect to a database

The Lakebase sink supports the following connection methods:

Lakebase tables registered with Unity Catalog

For Lakebase tables registered with Unity Catalog, the connector automatically manages the credentials and uses the identity of the user or service principal running the query. If the table doesn't exist, the connector creates the table.

To register a Lakebase database with Unity Catalog, see Register a Lakebase database in Unity Catalog.

To write to a Lakebase table, use the .toTable() method with a fully qualified table name, 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>")

Replace the following placeholders:

  • <catalog>.<schema>.<table>: The fully qualified name of the target table. The catalog is the Unity Catalog catalog you created when you registered the Lakebase database, see Register a Lakebase database in Unity Catalog. If the table doesn't exist, the connector creates it.
  • <primary-key-columns>: Optional. A comma-separated list of all columns in the target table's primary key, for example id or user_id,event_type. See Upsert behavior.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: A Unity Catalog volume path where the query stores its checkpoint. You can also use a cloud object storage URI. The location must be storage that you can write to, not local disk, and must be unique to each streaming query. This is independent of the target table. See Structured Streaming checkpoints.

For optional configurations, such as batchsize and batchinterval, see Configuration options.

Lakebase tables not registered with Unity Catalog

For Lakebase tables not registered with Unity Catalog, the connector automatically manages the credentials and uses the identity of the user or service principal running the query. If the table doesn't exist, the connector creates the table.

To write to a Lakebase table, use the endpoint and dbtable options:

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()

Replace the following placeholders:

  • <project-id>.<branch-id>.<endpoint-id>: Your Lakebase endpoint. Find all three values in the Resource name on the Get ID menu of the Computes tab, which has the format projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. See Compute identifiers.
  • <database>: Optional. The name of the target PostgreSQL database. Defaults to databricks_postgres. See Manage databases.
  • <schema>.<table>: The target table in schema.table format. If you omit the schema, the sink uses the public schema. For automatic table creation, use identifiers that start with a letter or underscore and contain only letters, numbers, and underscores.
  • <primary-key-columns>: Optional. A comma-separated list of all columns in the target table's primary key, for example id or user_id,event_type. See Upsert behavior.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: A Unity Catalog volume path where the query stores its checkpoint. You can also use a cloud object storage URI. The location must be storage that you can write to, not local disk, and must be unique to each streaming query. This is independent of the target table. See Structured Streaming checkpoints.

For optional configurations, such as batchsize and batchinterval, see Configuration options.

External PostgreSQL with Unity Catalog credentials

In Databricks Runtime 19 and above, use a Unity Catalog connection to authenticate to an external PostgreSQL database without storing credentials in your code. The target table must already exist.

Create a connection of type POSTGRESQL, see Create a connection. The user or service principal running the query must have USE CONNECTION on the connection.

To write to the PostgreSQL table, use the databricks.connection, database, and dbtable options:

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()

Replace the following placeholders:

  • <connection-name>: The name of the Unity Catalog connection.
  • <database>: The name of the target PostgreSQL database.
  • <schema>.<table>: The existing target table in schema.table format. If you omit the schema, the sink uses the public schema.
  • <primary-key-columns>: Optional. A comma-separated list of all columns in the target table's primary key, for example id or user_id,event_type. See Upsert behavior.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: A Unity Catalog volume path where the query stores its checkpoint. You can also use a cloud object storage URI. The location must be storage that you can write to, not local disk, and must be unique to each streaming query. This is independent of the target table. See Structured Streaming checkpoints.

PostgreSQL connections always use TLS. Certificate verification follows the settings on the Unity Catalog connection, which you choose when you create the connection:

  • Trust server certificate: When selected, the connection uses sslmode=require, which encrypts the connection without verifying the server certificate.
  • User provided server certificate: Provide a PEM-encoded server certificate to use sslmode=verify-full when Trust server certificate isn't selected. If you don't provide a certificate, the connection uses sslmode=verify-full with the JVM default trust store.

Configuration options

The sink raises an error for unrecognized options, JDBC_STREAMING_SINK_INVALID_OPTIONS.

The following options apply to all connection methods:

Key Default Description
batchinterval 100 milliseconds Optional. The maximum time to hold rows in the buffer before flushing. For example, "50 milliseconds".
batchsize 1000 Optional. The maximum number of rows for each database transaction.
checkpointLocation None Required. Path to a checkpoint directory, such as a Unity Catalog volume (/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>). Must be unique to each query. See Structured Streaming checkpoints.
upsertkey None Optional. A comma-separated list of all columns in the target table's primary key, for example, "id" or "user_id,event_type". See Upsert behavior.

Lakebase tables not registered with Unity Catalog

The following options apply when you connect to a Lakebase table not registered with Unity Catalog:

Key Default Description
database databricks_postgres Optional. The target PostgreSQL database name.
dbtable None Required. The target table name in schema.table format. If you don't specify a schema, the default schema value is public. For automatic table creation, use identifiers that start with a letter or underscore and contain only letters, numbers, and underscores.
endpoint None Required. The Lakebase endpoint, in project_id.branch_id or project_id.branch_id.endpoint_id format. The endpoint_id is optional. If you omit it and the branch has a single read-write endpoint, the sink selects that endpoint by default.

External PostgreSQL with Unity Catalog credentials

The following options apply when you connect to an external PostgreSQL database with Unity Catalog credentials:

Key Default Description
database None Required. The target PostgreSQL database name.
databricks.connection None Required. The Unity Catalog connection name for Unity Catalog-managed authentication to external PostgreSQL.
dbtable None Required. The existing target table name in schema.table format. If you don't specify a schema, the default schema value is public.

Data type mappings

The sink checks that each DataFrame column is compatible with its corresponding target column before writing to an existing Lakebase or external PostgreSQL table.

The following table contains types supported in Databricks Runtime 18 LTS and above:

Spark type Automatically created Lakebase table type Compatible types in existing PostgreSQL tables
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

The following table contains types supported in Databricks Runtime 19 and above:

Spark type Automatically created Lakebase table type Compatible types in existing PostgreSQL tables
DayTimeIntervalType, YearMonthIntervalType interval interval

Upsert behavior

The upsertkey option identifies the target table's primary key columns. For an existing table, the columns in upsertkey must match the table's primary key exactly. If you omit the option, the sink reads the primary key from the table. For a Lakebase table that the sink creates, upsertkey defines the primary key. If you omit the option, the sink creates the table without a primary key.

When the target table has a primary key, the sink upserts with PostgreSQL's INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... syntax. When the target table has no primary key, the sink performs inserts. A query's output mode has no effect on this behavior.

All primary key columns must be present in the DataFrame and use comparable types, such as numeric or string types.

Performance tuning

Batching and backpressure

A flush is triggered when either condition is met:

  • The buffer reaches batchsize rows, which defaults to 1000.
  • The buffer age exceeds batchinterval, which defaults to 100 milliseconds.

When the database cannot keep up with the incoming data rate, the sink propagates backpressure upstream to the source.

Latency and throughput guidance:

  • For low-latency workloads with real-time mode, decrease batchinterval to guarantee a shorter maximum time before flushing. See Real-time mode concepts for concepts and Real-time mode examples for a code example.
  • For high-throughput workloads, increase batchsize to reduce overhead for each transaction.

Connection behavior

The sink uses connection pooling on executors. By default, each task uses one database connection.

Databricks recommends that you use the default value of 1 task for each connection. If you increase the number of tasks for each connection, you might cause connection contentions and increase latencies for high throughput connections.

To configure the ratio of tasks to connections, set the spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark configuration. If the target database has a low connection limit, reduce the number of shuffle partitions or increase spark.databricks.sql.streaming.jdbc.tasksPerConnection.

The sink automatically retries transient JDBC errors, including connection failures, deadlocks, and rate limiting. If the sink exhausts all retries, the query fails.

Supported triggers and output modes

Triggers

This table shows support for Structured Streaming trigger types on classic and serverless compute:

Trigger Classic compute Serverless compute (notebooks and jobs)
RealTime Yes No
ProcessingTime Yes No
AvailableNow Yes Yes
Once Yes. Deprecated. Use AvailableNow. Yes. Deprecated. Use AvailableNow.

Output modes

This table shows support for Structured Streaming output modes:

Output mode Supported
update Yes
append Yes. Behavior is identical to update. The query upserts when the target table has a primary key, otherwise the query inserts. See Upsert behavior.
complete No

Limitations

  • For an external PostgreSQL database connected through a Unity Catalog connection, the target table must already exist. The sink automatically creates missing tables only in Lakebase.
  • Lakeflow pipelines are not supported.
  • Last updated on