Подключение к Lakebase

Используйте Structured Streaming для записи в Lakebase или внешнюю базу данных PostgreSQL с встроенным пакетированием, автоматическими повторами и аутентификацией, управляемой рабочим пространством.

Когда следует использовать приемник Lakebase

Используйте приемник Lakebase для потоковой записи с низкой задержкой в Lakebase или во внешнюю базу данных PostgreSQL. Этот приемник не требует, чтобы вы реализовывали пользовательские foreach функции для пакетной обработки, управления подключениями и обработки ошибок.

Распространенные варианты использования:

Обновите базы данных приложений в режиме реального времени для операционных панелей мониторинга или функций, доступных для клиентов.
Синхронизация непрерывно изменяющихся данных, таких как агрегированные или отфильтрованные результаты потоковой передачи, в базу данных транзакций.
Записывайте результаты запроса Structured Streaming в таблицу Lakebase с задержкой менее секунды с помощью режима реального времени.

Чтобы синхронизировать данные из Lakebase с таблицами Delta Lake в Lakehouse, о синхронизации в обратном направлении см. канал передачи данных об изменениях Lakebase.

Требования

Databricks Runtime 18 LTS и выше.
- Для внешних подключений PostgreSQL необходимо использовать Databricks Runtime 19 или более поздней версии, а также включить предварительную версию Custom JDBC on UC Compute.
- Для интервальных типов данных требуется использование Databricks Runtime 19 и выше.
- Стратегии слияния для upsert требуют использования Databricks Runtime 19 и выше.
Классические вычисления с выделенными или стандартными режимами доступа, или бессерверные вычисления для ноутбуков или задач. Для бессерверных вычислений используйте Trigger.AvailableNow(). См. Потоковая передача данных в бессерверных вычислениях.
База данных Lakebase или соединение Unity Catalog с внешней базой данных PostgreSQL.

Требования к идентификатору

Для всех целей Databricks рекомендует использовать схемы, таблицы, столбцы и основные ключевые названия столбцов, которые начинаются с буквы или подчеркивания и содержат только буквы, цифры и подчёркивания. При автоматическом создании таблицы Lakebase приёмник обеспечивает выполнение этих требований. Чтобы использовать идентификаторы, не соответствующие этим требованиям, создайте целевой таблицу перед началом запроса.

Соединение с базой данных

Приемник Lakebase поддерживает следующие методы подключения:

Таблицы Lakebase, зарегистрированные в каталоге Unity

Для таблиц Lakebase, зарегистрированных в Unity Catalog, коннектор автоматически управляет учетными данными и использует идентификационные данные пользователя или сервисного субъекта, выполняющего запрос. Если таблица не существует, соединитель создает таблицу.

Сведения о регистрации базы данных Lakebase в каталоге Unity см. в разделе "Регистрация базы данных Lakebase" в каталоге Unity.

Для записи в таблицу 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 каталог каталога Unity, созданный при регистрации базы данных Lakebase, см. статью "Регистрация базы данных Lakebase в каталоге Unity". Если таблица не существует, соединитель создает ее.
<primary-key-columns>: необязательный параметр. Список всех столбцов в первичном ключе целевой таблицы, например id или user_id,event_type. См. поведение Апсерта.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Путь тома каталога Unity, в котором запрос сохраняет контрольную точку. Вы также можете использовать универсальный код ресурса (URI) хранилища облачных объектов. Расположение должно быть хранилищем, которое можно записать, а не локальный диск, и должно быть уникальным для каждого потокового запроса. Это не зависит от целевой таблицы. Смотрите контрольные точки структурированной потоковой передачи.

Для необязательных конфигураций, таких как batchsize и batchinterval, см. параметры приемника PostgreSQL.

Таблицы Lakebase, не зарегистрированные в каталоге Unity

Для таблиц Lakebase, не зарегистрированных в каталоге Unity, соединитель автоматически управляет учетными данными и использует удостоверение пользователя или субъекта-службы, выполняющего запрос. Если таблица не существует, соединитель создает таблицу.

Чтобы записывать данные в таблицу 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. Найдите все три значения в имени ресурса в меню "Получить идентификатор " вкладки "Вычисления ", которая имеет формат 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. См. поведение Апсерта.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Путь тома каталога Unity, в котором запрос сохраняет контрольную точку. Вы также можете использовать универсальный код ресурса (URI) хранилища облачных объектов. Расположение должно быть хранилищем, которое можно записать, а не локальный диск, и должно быть уникальным для каждого потокового запроса. Это не зависит от целевой таблицы. Смотрите контрольные точки структурированной потоковой передачи.

Для необязательных конфигураций, таких как batchsize и batchinterval, см. параметры приемника PostgreSQL.

Внешний PostgreSQL с учетными данными каталога Unity

Important

Эта функция доступна в общедоступной предварительной версии. Администраторы рабочих пространств могут управлять доступом к пользовательскому JDBC на UC Compute с страницы Preview . См. статью "Управление предварительными версиями 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 Catalog.
<database>: Название целевой базы данных PostgreSQL.
<schema>.<table>: Существующая целевая таблица в schema.table формате. Если вы не укажете схему, приемник использует схему public.
<primary-key-columns>: необязательный параметр. Список всех столбцов в первичном ключе целевой таблицы, например id или user_id,event_type. См. поведение Апсерта.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Путь тома каталога Unity, в котором запрос сохраняет контрольную точку. Вы также можете использовать универсальный код ресурса (URI) хранилища облачных объектов. Расположение должно быть хранилищем, которое можно записать, а не локальный диск, и должно быть уникальным для каждого потокового запроса. Это не зависит от целевой таблицы. Смотрите контрольные точки структурированной потоковой передачи.

Подключения PostgreSQL всегда используют TLS. Проверка сертификатов следует настройкам соединения Unity Catalog, которое вы выбираете при создании соединения:

Сертификат доверенного сервера: При выборе соединение использует sslmode=require, который шифрует соединение без проверки сертификата сервера.
Серверный сертификат, предоставленный пользователем: Укажите серверный сертификат в формате PEM для использования sslmode=verify-full , если параметр Доверять сертификату сервера не выбран. Если вы не предоставляете сертификат, соединение используется sslmode=verify-full с стандартным хранилищем доверия JVM.

Параметры конфигурации

Приёмник сообщает об ошибке при использовании нераспознанных параметров, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Параметры конфигурации приемника, включая общие параметры и параметры для каждого метода подключения, см. в разделе параметры приемника PostgreSQL.

Сопоставления типов данных

Приёмник проверяет, совместим ли каждый столбец в DataFrame с соответствующим целевым столбцом, прежде чем записывать данные в существующую таблицу Lakebase или внешнюю таблицу PostgreSQL.

Следующая таблица содержит типы, поддерживаемые в 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`

Поведение при обновлении или вставке

Эта upsertkey опция определяет основные ключевые столбцы целевой таблицы. Для существующей таблицы столбцы в upsertkey должны точно совпадать с первичным ключом таблицы. Если вы не включите эту опцию, раковина считывает первичный ключ из таблицы. Для таблицы Lakebase, создаваемой приёмником, upsertkey определяет первичный ключ. Если не указать этот параметр, приёмник создаст таблицу без первичного ключа.

Когда целевая таблица имеет первичный ключ, приёмник выполняет UPSERT с использованием синтаксиса PostgreSQL INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET .... Когда целевая таблица не имеет первичного ключа, приемник выполняет операции вставки. Режим вывода запроса не влияет на это поведение.

Все столбцы первичных ключей должны присутствовать в DataFrame и использовать сопоставимые типы, такие как числовые или строки.

Чтобы контролировать, как операции upsert обновляют столбцы, не входящие в первичный ключ, см. стратегии слияния для операций upsert.

Стратегии слияния для операций upsert

В Databricks Runtime 19 и выше вы можете использовать стратегии слияния, чтобы контролировать, как приемник обновляет столбцы с непервичным ключом, если строки имеют одинаковый первичный ключ. Например, вы можете сохранить наибольшую временную метку или обновить столбец только тогда, когда входящее событие имеет более новое значение сортировки.

Стратегии слияния применяются только к upsert. Если целевая таблица не имеет первичного ключа, настройка upsert.mergeStrategies или upsert.defaultMergeStrategy вызывает ошибку. Стратегии можно назначать только столбцам, не являющимся первичным ключом. Установите upsert.mergeStrategies в значение списка элементов, разделённых запятыми, в формате column:strategy. Установите upsert.defaultMergeStrategy значение по умолчанию для столбцов без явных стратегий. upsert.defaultMergeStrategy не принимает ни latestBy, ни earliestBy. По умолчанию приемник использует эту latest стратегию.

Поддерживаемые стратегии слияния

Strategy	Behavior
`latest`	Используйте последнее значение в порядке поступления. Это значение по умолчанию.
`earliest`	Используйте самое раннее значение по порядку прибытия.
`latestNonNull`	Используйте последнее ненулевое значение по порядку прибытия.
`earliestNonNull`	Используйте самое раннее ненулево значение по порядку прибытия.
`greatest`	Используйте большее из существующих и входящих значений. Нулевые значения игнорируются, поэтому нулевое входящее значение не заменяет существующее.
`least`	Используйте меньшее из существующего и входящего значений. Нулевые значения игнорируются, поэтому нулевое входящее значение не заменяет существующее.
`latestBy(column)`	Используйте значение из строки с наибольшим значением в указанном столбце сортировки. Смотрите, как `latestBy` и `earliestBy` обрабатывают совпадения и значения NULL.
`earliestBy(column)`	Используйте значение из строки с наименьшим значением в указанном столбце сортировки. См., как `latestBy` и `earliestBy` обрабатывают равные значения и NULL.

Как `earliestBy` и `latestBy` обрабатывают ничьи и значения NULL

latestBy(column) и earliestBy(column) выбирают значение, сравнивая входящее значение порядка с тем, что уже хранится в таблице. Чтобы сохранить точность сохраненного значения, столбец упорядочения должен использовать стратегию сопоставления, либо в upsert.mergeStrategies , либо через upsert.defaultMergeStrategy:

Для latestBy(column), столбец упорядочения должен использовать greatest.
Для earliestBy(column), столбец упорядочения должен использовать least.

Например, для is_success:latestBy(last_updated_at) требуется last_updated_at:greatest.

Когда оба значения порядка не равны нулю, latestBy(column) обновляет столбец только тогда, когда входящее значение порядка строго больше (>) существующего значения порядка. Для earliestBy(column) должно быть строго меньше (<). Для обеих стратегий, когда значения сортировки равны, приемник сохраняет существующее значение столбца. В одной партии, если несколько строк имеют одинаковый первичный ключ и значение порядка, приемник сохраняет первую полученную строку.

Например, если is_success использует latestBy(last_updated_at) и входящая строка имеет то же самое last_updated_at, что и существующая, то приемник сохраняет существующее is_success, даже если входящее значение отличается.

Для обеих стратегий ненулевые значения упорядочения имеют приоритет над нулевыми значениями. Входящая строка с ненулевым порядковым значением заменяет существующее значение, у которого порядковое значение равно null. Входящая строка со значением порядка null не заменяет существующее значение, включая случаи, когда существующее значение порядка также имеет значение null.

Пример: Пользовательская логика upsert модели

Рассмотрим таблицу с столбцами id, last_updated_at, creation_date, и is_success. Возможно, стоит выразить следующую логику:

INSERT INTO events ...
ON CONFLICT (id) DO UPDATE
SET
  -- Keep the most recent timestamp.
  last_updated_at = GREATEST(events.last_updated_at, EXCLUDED.last_updated_at),

  -- Keep the earliest non-null creation date.
  creation_date = COALESCE(events.creation_date, EXCLUDED.creation_date),

  -- Keep the value of is_success from the event with the most recent last_updated_at.
  is_success = CASE
    WHEN EXCLUDED.last_updated_at IS NOT NULL
      AND (events.last_updated_at IS NULL
        OR EXCLUDED.last_updated_at > events.last_updated_at)
    THEN EXCLUDED.is_success
    ELSE events.is_success
  END;

Для моделирования этой логики используйте следующие стратегии слияния:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "id")
  .option("upsert.mergeStrategies", "last_updated_at:greatest,creation_date:earliestNonNull,is_success:latestBy(last_updated_at)")
  .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", "id")
  .option("upsert.mergeStrategies", "last_updated_at:greatest,creation_date:earliestNonNull,is_success:latestBy(last_updated_at)")
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Настройка производительности

Пакетная обработка и обратное давление

При выполнении любого условия выполняется очистка:

Размер буфера достигает batchsize строк, значение по умолчанию — 1000.
Возраст буфера превышает batchinterval, которое по умолчанию равно 100 milliseconds.

Если база данных не справляется со скоростью поступления данных, приёмник передаёт обратное давление вверх по потоку к источнику.

Руководство по задержке и пропускной способности:

Для рабочих нагрузок с низкой задержкой, использующих режим реального времени, уменьшите batchinterval, чтобы гарантировать меньшее максимальное время до сброса. См. концепции режима реального времени для концепций и примеры режима реального времени для примера кода.
Для рабочих нагрузок с высокой пропускной способностью увеличьте batchsize, чтобы уменьшить накладные расходы на каждую транзакцию.

Поведение подключения

Приемник использует пул соединений для исполнителей. По умолчанию каждая задача использует одно подключение к базе данных.

Databricks рекомендует использовать значение 1 задачи по умолчанию для каждого подключения. Если увеличить количество задач для каждого подключения, это может привести к конфликтам на соединении и увеличить задержки для соединений с высокой пропускной способностью.

Чтобы настроить отношение задач к подключениям, задайте конфигурацию spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark. Если для целевой базы данных установлен низкий лимит подключений, уменьшите количество разделов shuffle или увеличьте spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Приемник автоматически повторяет попытки при возникновении временных ошибок JDBC, включая сбои подключения, взаимоблокировки и ограничение частоты запросов. Если приемник исчерпает все повторные попытки, запрос завершается ошибкой.

Поддерживаемые триггеры и режимы вывода

Triggers

В этой таблице показана поддержка типов триггеров структурированного потокового потока на классических и бессерверных вычислениях:

Триггер	Классические вычисления	Бессерверные вычисления (ноутбуки и задания)
`RealTime`	Yes	No
`ProcessingTime`	Yes	No
`AvailableNow`	Yes	Yes
`Once`	Yes. Deprecated. Используйте `AvailableNow`.	Yes. Deprecated. Используйте `AvailableNow`.

Режимы вывода

В этой таблице показана поддержка режимов вывода структурированной потоковой передачи:

Режим вывода	Supported
`update`	Yes
`append`	Yes. Поведение идентично `update`. Запрос выполняет операцию upsert, если целевая таблица имеет первичный ключ; в противном случае выполняется вставка. См. поведение Апсерта.
`complete`	No

Ограничения

Для внешней базы данных PostgreSQL, подключённой через Unity Catalog, целевая таблица должна уже существовать. Приёмник автоматически создаёт недостающие таблицы только в Lakebase.
Трубопроводы Lakeflow не поддерживаются.

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-11-04

Подключение к Lakebase

Когда следует использовать приемник Lakebase

Требования

Требования к идентификатору

Соединение с базой данных

Таблицы Lakebase, зарегистрированные в каталоге Unity

Python

Scala

Таблицы Lakebase, не зарегистрированные в каталоге Unity

Python

Scala

Внешний PostgreSQL с учетными данными каталога Unity

Python

Scala

Параметры конфигурации

Сопоставления типов данных

Поведение при обновлении или вставке

Стратегии слияния для операций upsert

Поддерживаемые стратегии слияния

Как earliestBy и latestBy обрабатывают ничьи и значения NULL

Пример: Пользовательская логика upsert модели

Python

Scala

Настройка производительности

Пакетная обработка и обратное давление

Поведение подключения

Поддерживаемые триггеры и режимы вывода

Triggers

Режимы вывода

Ограничения

Обратная связь

Дополнительные ресурсы

Как `earliestBy` и `latestBy` обрабатывают ничьи и значения NULL