Добавьте Базу данных PostgreSQL CDC как источник в узел Real-Time

В этой статье описывается, как добавить отслеживание изменений данных (CDC) базы данных PostgreSQL в качестве источника событий в узле Real-Time Fabric.

Коннектор источника для отслеживания изменений данных (CDC) в потоках событий Microsoft Fabric позволяет захватывать моментальный снимок текущих данных в базе данных PostgreSQL. В настоящее время служба PostgreSQL Database Change Data Capture (CDC) поддерживается из следующих служб, где базы данных могут быть доступны для общедоступного доступа:

• База данных Azure для PostgreSQL
• Amazon RDS для PostgreSQL
• Amazon Aurora PostgreSQL
• Google Cloud SQL для PostgreSQL

После добавления источника CDC базы данных PostgreSQL в поток событий он записывает изменения на уровне строк в указанные таблицы. Затем эти изменения можно обрабатывать в режиме реального времени и отправлять в разные места назначения для дальнейшего анализа.

Примечание

С помощью DeltaFlow (предварительная версия) можно преобразовать необработанные события Debezium CDC в потоки, готовые к аналитике, которые отражают структуру исходной таблицы. DeltaFlow автоматизирует регистрацию схемы, управление таблицами назначения и обработку эволюции схемы. Чтобы использовать DeltaFlow, выберите события, готовые к аналитике, и автоматически обновленную схему на этапе обработки схемы.

Prerequisites

• Доступ к рабочей области в режиме лицензии на емкость Fabric или в режиме пробной лицензии с разрешениями Контрибьютора или более высокого уровня.

• Зарегистрированный доступ пользователей в базе данных PostgreSQL.

• База данных PostgreSQL должна быть общедоступной, а не за брандмауэром или защищенной в виртуальной сети. Если он находится в защищенной сети, подключитесь к нему с помощью инъекции виртуальной сети соединителя Eventstream.

• CDC включен в базе данных и таблицах PostgreSQL.

  Если у вас есть База данных Azure для PostgreSQL, выполните действия, описанные в следующем разделе, чтобы включить CDC. Подробные сведения см. в разделе "Логическая репликация и логическое декодирование" — База данных Azure для PostgreSQL — гибкий сервер.

  Для других баз данных PostgreSQL см. Debezium connector for PostgreSQL :: документация Debezium.

• Если у вас нет потока событий, создайте поток событий.

Включение CDC в базе данных PostgreSQL

В этом разделе в качестве примера используется База данных Azure для PostgreSQL.

Чтобы включить CDC в Azure Database for PostgreSQL Flexible Server, выполните следующие действия.

1. На странице Azure Database for PostgreSQL Flexible Server в портале Azure выберите параметры сервера в меню навигации.

2. На странице параметров сервера:

   • Установите wal_level на логический.
   • Обновите max_worker_processes до значения не менее 16.

   

3. Сохраните изменения и перезапустите сервер.

4. Убедитесь, что экземпляр гибкого сервера Azure Database for PostgreSQL разрешает доступ к общедоступному сетевому трафику.

5. Предоставьте администраторам разрешения на репликацию, выполнив следующую инструкцию SQL. Если вы хотите использовать другую учетную запись пользователя для подключения базы данных PostgreSQL (DB) для получения CDC, убедитесь, что пользователь является владельцем таблицы.

   ALTER ROLE <admin_user_or_table_owner_user> WITH REPLICATION;
   

Перейдите на страницу источников данных

1. Войдите в Microsoft Fabric.

2. Если вы видите Power BI в нижней левой части страницы, переключитесь на рабочую нагрузку Fabric, выбрав Power BI, а затем выбрав Fabric.

   

3. Выберите режим реального времени на левой панели навигации.

   

4. Страница потоковых данных открывается по умолчанию. Нажмите кнопку "Добавить данные ", чтобы перейти на страницу источников данных .

   

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

   

Выбор базы данных PostgreSQL CDC в качестве типа источника

На странице источников данных выберите категорию источников Майкрософт в верхней части и щелкните "Подключиться" на плитке Azure DB для PostgreSQL (CDC).

Настройка источника данных CDC в базе данных Azure для PostgreSQL

Прием изменений данных из баз данных PostgreSQL с автоматической регистрацией схем таблиц через функцию CDC в Eventstream.

Примечание

DeltaFlow (предварительная версия): при выборе событий, готовых к аналитике, и автоматического обновления схемы на этапе обработки схемы, DeltaFlow преобразует необработанные события Debezium CDC в потоки, готовые к аналитике, которые отражают структуру исходной таблицы. DeltaFlow также автоматизирует создание целевой таблицы и обработку эволюции схемы.

1. На странице "Подключение" выберите "Создать подключение".

   

2. В разделе "Параметры подключения" введите следующие сведения.

   • Сервер: адрес сервера базы данных PostgreSQL, например my-pgsql-server.postgres.database.azure.com.

   • База данных: имя базы данных, например my_database.

     

   • Имя подключения: введите имя подключения.

   • Тип проверки подлинности, выберите "Базовый" и введите имя пользователя и пароль для базы данных.

     Примечание

     В настоящее время потоки событий Fabric поддерживают только обычную проверку подлинности.

   • Выберите "Подключиться" , чтобы завершить параметры подключения.

3. Порт. Введите номер порта сервера. Значение по умолчанию — 5432. Если выбранное облачное подключение настроено в Управление подключениями и шлюзами, убедитесь, что номер порта соответствует тому, который указан там. Если они не совпадают, то номер порта облачного подключения в Управление подключениями и шлюзами имеет приоритет.

4. При записи изменений из таблиц базы данных можно выбрать один из двух вариантов:

   • Все таблицы: запись изменений из каждой таблицы в базе данных.
   • Введите имена таблиц: позволяет указать подмножество таблиц с помощью списка, разделенного запятыми. Можно использовать полные идентификаторы таблицы в формате schemaName.tableName или допустимых регулярных выражениях. Примеры:
   • dbo.test.*: выберите все таблицы из схемы test, имена которых начинаются с dbo.
   • dbo\.(test1|test2): выберите dbo.test1 и dbo.test2.

   В списке можно объединить оба формата. Общее ограничение символов для всей записи составляет 102 400 символов.

5. Имя слота (необязательно): Введите имя слота логического декодирования PostgreSQL, который был создан для потоковой передачи изменений из определенного подключаемого модуля для конкретной базы данных или схемы. Сервер использует этот слот для потоковой передачи событий в соединитель потоковой передачи Eventstream. Он должен содержать только строчные буквы, цифры и знаки подчеркивания.

   • Если это не указано, для создания слота используется GUID, требующий соответствующих разрешений базы данных.
   • Если указанное имя слота существует, соединитель использует его напрямую.

6. Разверните расширенные настройки, чтобы получить доступ к дополнительным вариантам настройки источника CDC базы данных PostgreSQL.

   • Имя публикации: указывает имя публикации логической репликации PostgreSQL, используемой. Это значение должно соответствовать существующей публикации в базе данных или автоматически создаваться в зависимости от режима автоматического создания. Значение по умолчанию: dbz_publication.

     Примечание

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

   • Режим автоматического создания публикации: определяет, создается ли публикация и как она создается автоматически. Возможные варианты:

     • Filtered (по умолчанию): если указанная публикация не существует, соединитель создает публикацию, которая включает только выбранные таблицы (как указано в списке включаемых таблиц).
     • AllTables: Если указанная публикация существует, соединитель использует её. Если он не существует, коннектор создаёт объект, который включает в себя все таблицы в базе данных.
     • Disabled: соединитель не создает публикацию. Если указанная публикация отсутствует, соединитель создает исключение и останавливается. В этом случае публикация должна быть создана вручную в базе данных.

     Дополнительные сведения см. в документации Debezium по режиму автоматического создания публикации

   • Режим десятичной обработки: указывает, как соединитель обрабатывает значения PostgreSQL DECIMAL и NUMERIC столбцов:

     • Precise: представляет значения, использующие точные десятичные типы (например, Java BigDecimal), чтобы обеспечить полную точность и точность представления данных.
     • Double: преобразует значения в числа с плавающей запятой двойной точности. Этот параметр повышает удобство использования и производительность, но может привести к потере точности.
     • String: кодирует значения в виде форматированных строк. Этот параметр упрощает их использование в подчиненных системах, но теряет семантические сведения о исходном числовом типе.

   • Режим моментального снимка: укажите условия для выполнения моментального снимка при запуске соединителя:

     • Initial: Соединитель запускает снимок только если не записаны смещения для имени логического сервера или если обнаруживается, что предыдущий снимок не завершился. После завершения моментального снимка соединитель начинает потоковую передачу записей событий для последующих изменений базы данных.
     • InitialOnly: Соединитель запускает моментальный снимок только в том случае, если не зафиксированы сдвиги для имени логического сервера. После завершения моментального снимка соединитель останавливается. Он не переходит на потоковую передачу для чтения событий изменения из binlog.
     • NoData: соединитель запускает снапшот, который сохраняет только схему, но не данные таблицы. Установите этот параметр, если вам не нужен согласованный моментальный снимок данных, но необходимо только изменения, происходящие с момента запуска соединителя.

   • Запрос действия Heartbeat: указывает запрос, который соединитель выполняет в исходной базе данных, когда соединитель отправляет сообщение пульса.

   • Переопределение инструкции выбора снимка: указывает строки таблицы, которые необходимо включить в снимок. Используйте свойство, если требуется, чтобы в снимок вошло только подмножество строк таблицы. Это свойство влияет только на моментальные снимки. Он не применяется к событиям, которые соединитель считывает из журнала.

Сведения о потоке или источнике

1. На странице "Подключение" выполните одно из этих действий на основе того, используете ли вы поток событий или концентратор Real-Time.

   • Eventstream:

     В области сведений о источнике справа выполните следующие действия:

     1. Чтобы изменить имя источника, выберите иконку «Карандаш».

     2. Обратите внимание, что имя потока событий и имя Stream доступны только для чтения.

   • центр реального времени:

     В разделе сведений о потоке справа выполните следующие действия.

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

     2. Для потока событий нажмите кнопку "Карандаш" и введите имя для потока событий.

     3. Значение имени потока автоматически создается шляхом добавления -stream к имени eventstream. Этот поток отображается на странице "Все потоки данных " концентратора в режиме реального времени, когда мастер завершит работу.

2. Нажмите кнопку "Далее" в нижней части страницы "Настройка ".

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

На экране "Обзор и подключение" просмотрите сводку и выберите "Добавить" (Eventstream) или "Подключить" (Real-Time hub).

Страница обработки схемы

1. На шаге обработки схемы выберите один из следующих вариантов:

   • События, готовые к аналитике и автоматически обновляемая схема (предварительная версия DeltaFlow): соединитель преобразует необработанные события CDC в потоки, готовые к аналитике, которые отражают структуру исходной таблицы. DeltaFlow дополняет события метаданными, такими как тип изменения (вставка, обновление или удаление) и метки времени, а также автоматически управляет целевыми таблицами и эволюцией схемы.
   • Необработанные события CDC: Соединитель принимает и делает доступными необработанные события CDC. При необходимости соединитель может автоматически обнаружить схемы таблиц и зарегистрировать их в реестре схем. Используйте этот параметр, если требуется осведомленность о схеме без преобразования DeltaFlow.

   Примечание

   На следующем скриншоте показан CDC для Azure SQL Database. Параметры обработки схем одинаковы для всех поддерживаемых соединителей источника CDC.

   

2. Включите ассоциацию схемы событий.

3. Для рабочей области выберите рабочую область Fabric для набора схем.

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

5. Если вы выбрали параметр +Создать на предыдущем шаге, введите имя набора схем.

6. На странице «Проверка + подключение» просмотрите сводку, а затем выберите «Добавить» (Eventstream) или «Подключить» (Real-Time hub).

   

   Для всех таблиц или выбранных таблиц в базе данных PostgreSQL соединитель автообнаружает и создает схемы и регистрирует их в реестре схем.

DeltaFlow: готовое для аналитики преобразование событий (предварительная версия)

При включении событий, готовых для аналитики, и автообновляемой схемы (DeltaFlow) коннектор предоставляет следующие возможности:

• Фигура события, готовая к аналитике: Сырые события Debezium CDC преобразуются в табличный формат, который отражает структуру исходной таблицы. События обогащены столбцами метаданных, включая тип изменения (insertилиupdatedelete) и метку времени события.
• Автоматическое управление таблицами назначения: при маршрутизации потоков с поддержкой DeltaFlow в поддерживаемое назначение, например в хранилище событий, целевые таблицы автоматически создаются для сопоставления схемы исходной таблицы. Вам не нужно вручную создавать или настраивать целевые таблицы.
• Обработка эволюции схемы. При изменении исходных таблиц баз данных (например, добавляются новые столбцы или создаются таблицы), DeltaFlow автоматически обнаруживает изменения, обновляет зарегистрированные схемы и корректирует целевые таблицы соответствующим образом. Это поведение сводит к минимуму вмешательство вручную, вызванное изменениями схемы.

Примечание

DeltaFlow (предварительная версия) в настоящее время поддерживается для работы с CDC в базе данных SQL в Azure, CDC для управляемого экземпляра SQL Azure, CDC для SQL Server на виртуальных машинах и источниковыми коннекторами CDC для PostgreSQL.

Дополнительные сведения о том, как DeltaFlow преобразует необработанные события CDC в готовые к аналитике выходные данные, включая типы операций и столбцы метаданных, см. в разделе "Преобразование выходных данных DeltaFlow".

Просмотр сведений о потоке данных

1. На странице "Проверка и подключение " при выборе "Открыть поток событий" мастер открывает поток событий, созданный для вас с выбранной базой данных PostgreSQL CDC в качестве источника. Чтобы закрыть мастер, нажмите кнопку Готово в нижней части страницы.

   

2. Поток должен отображаться в разделе "Последние потоковые данные" на главной странице центраReal-Time. Подробные инструкции см. в разделе «Просмотр сведений о потоках данных в Fabric Real-Time Hub».

Связанный контент

Дополнительные сведения об использовании потоков данных см. в следующих статьях:

• Обработка потоков данных
• Анализ потоков данных
• Настройка оповещений в потоках данных