Skip to main content
Last updated on

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.

Requirements​

  • Databricks Runtime 18 LTS and above.
    • External PostgreSQL connections require you to use Databricks Runtime 19 and above and to opt in to the Custom JDBC on UC Compute preview.
    • Interval data types require you to use Databricks Runtime 19 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 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 Databricks 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()
)

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 the PostgreSQL sink options.

External PostgreSQL with Unity Catalog credentials​

Preview

This feature is in Public Preview. Workspace admins can control access to Custom JDBC on UC Compute from the Previews page. See Manage Databricks previews.

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

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.

For the sink configuration options, including the common options and the options for each connection method, see the PostgreSQL sink options.

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

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

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.

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

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.
On this page