Сбор данных изменений (CDC) со встроенным Debezium и Spring Boot

При работе с данными или репликации источников данных вы, вероятно, слышали термин « сбор данных изменений» (CDC) . Как следует из названия, «CDC» — это шаблон проектирования, который постоянно идентифицирует и фиксирует постепенные изменения данных. Этот шаблон используется для репликации данных в реальном времени через действующие базы данных в аналитические источники данных или для чтения реплик. Он также может использоваться для запуска событий на основе изменений данных, таких как шаблон OutBox .

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

В этом блоге я хотел сосредоточиться на использовании фреймворка, обычно используемого для CDC , и встраивании его в Spring Boot.

Вам также могут понравиться: Коннекторы Kafka Без Kafka .

Что такое дебезиум?

Debezium — это распределенная платформа, созданная для CDC. Он использует журналы транзакций базы данных и создает потоки событий при изменениях на уровне строк. Приложения, прослушивающие эти события, могут выполнять необходимые действия на основе постепенных изменений данных.

Debezium предоставляет библиотеку соединителей , поддерживающих различные базы данных, доступные сегодня. Эти соединители могут отслеживать и записывать изменения на уровне строк в схемах базы данных. Затем они публикуют изменения в потоковом сервисе, таком как Kafka .

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

Что такое встраиваемый дебезиум?

Приложения, которым не требуется уровень отказоустойчивости и надежности, которые предлагает Kafka Connect, или которые хотят минимизировать затраты на их использование для работы всей платформы, могут использовать в приложении соединители Debezium. Это делается путем встраивания движка Debezium и настройки соединителя для работы в приложении. При событиях изменения данных соединители отправляют их непосредственно в приложение.

Дебезиум с пружинным ботинком

Для простоты примера давайте создадим приложение Spring Boot «Student CDC Relay», в котором будет запущен встроенный Debezium и настроены журналы транзакций базы данных Postgres, в которой находится таблица «Student». Коннектор Debezium, настроенный в приложении Spring Boot, вызывает метод в приложении, когда в таблице «Студент» выполняется операция базы данных, такая как «Вставить / Обновить / Удалить». Метод воздействует на эти события и синхронизирует данные в индексе Student в ElasticSearch.

Установка инструментов

Все необходимые инструменты могут быть установлены с помощью файла docker-compose ниже. Это запустит базу данных Postgres через порт  5432 и Elastic Search через порт 9200 (HTTP)  и  9300 (транспорт) .

version: "3.5"

services:
  # Install postgres and setup the student database.
  postgres:
    container_name: postgres
    image: debezium/postgres
    ports:
      - 5432:5432
    environment:
      - POSTGRES_DB=studentdb
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password

  # Install Elasticsearch.
  elasticsearch:
    container_name: elasticsearch
    image: docker.elastic.co/elasticsearch/elasticsearch:6.8.0
    environment:
    - discovery.type=single-node
    ports:
      - 9200:9200
      - 9300:9300

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>${debezium.version}</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-postgres</artifactId>
    <version>${debezium.version}</version>
</dependency>

@Bean
public io.debezium.config.Configuration studentConnector() {
    return io.debezium.config.Configuration.create()
            .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
            .with("offset.storage",  "org.apache.kafka.connect.storage.FileOffsetBackingStore")
            .with("offset.storage.file.filename", "/path/cdc/offset/student-offset.dat")
            .with("offset.flush.interval.ms", 60000)
            .with("name", "student-postgres-connector")
            .with("database.server.name", studentDBHost+"-"+studentDBName)
            .with("database.hostname", studentDBHost)
            .with("database.port", studentDBPort)
            .with("database.user", studentDBUserName)
            .with("database.password", studentDBPassword)
            .with("database.dbname", studentDBName)
            .with("table.whitelist", STUDENT_TABLE_NAME).build();
}

private CDCListener(Configuration studentConnector, StudentService studentService) {
    this.engine = EmbeddedEngine
            .create()
            .using(studentConnector)
            .notifying(this::handleEvent).build();

    this.studentService = studentService;
}

    private final Executor executor = Executors.newSingleThreadExecutor();

    @PostConstruct
    private void start() {
        this.executor.execute(engine);
    }

    @PreDestroy
    private void stop() {
        if (this.engine != null) {
            this.engine.stop();
        }
    }

CREATE TABLE public.student
(
    id integer NOT NULL,
    address character varying(255),
    email character varying(255),
    name character varying(255),
    CONSTRAINT student_pkey PRIMARY KEY (id)
);

INSERT INTO STUDENT(ID, NAME, ADDRESS, EMAIL) VALUES('1','Jack','Dallas, TX','jack@gmail.com');

$ curl -X GET http://localhost:9200/student/student/1?pretty=true
{
  "_index" : "student",
  "_type" : "student",
  "_id" : "1",
  "_version" : 31,
  "_seq_no" : 30,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : 1,
    "name" : "Jack",
    "address" : "Dallas, TX",
    "email" : "jack@gmail.com"
  }
}

UPDATE STUDENT SET EMAIL='jill@gmail.com', NAME='Jill' WHERE ID = 1; 

$ curl -X GET http://localhost:9200/student/student/1?pretty=true
{
  "_index" : "student",
  "_type" : "student",
  "_id" : "1",
  "_version" : 32,
  "_seq_no" : 31,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : 1,
    "name" : "Jill",
    "address" : "Dallas, TX",
    "email" : "jill@gmail.com"
  }
}

DELETE FROM STUDENT WHERE ID = 1;

$ curl -X GET http://localhost:9200/student/student/1?pretty=true
{
  "_index" : "student",
  "_type" : "student",
  "_id" : "1",
  "_version" : 33,
  "_seq_no" : 32,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : 1,
    "name" : null,
    "address" : null,
    "email" : null
  }
}

ERROR 59453 --- [pool-2-thread-1] io.debezium.embedded.EmbeddedEngine      : Error while trying to run connector class 'io.debezium.connector.postgresql.PostgresConnector'

Caused by: org.postgresql.util.PSQLException: ERROR: replication slot "debezium" is active for PID <>
at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2440) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1116) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.core.v3.QueryExecutorImpl.startCopy(QueryExecutorImpl.java:842) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.core.v3.replication.V3ReplicationProtocol.initializeReplication(V3ReplicationProtocol.java:58) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.core.v3.replication.V3ReplicationProtocol.startLogical(V3ReplicationProtocol.java:42) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.replication.fluent.ReplicationStreamBuilder$1.start(ReplicationStreamBuilder.java:38) ~[postgresql-42.2.5.jar:42.2.5]
at org.postgresql.replication.fluent.logical.LogicalStreamBuilder.start(LogicalStreamBuilder.java:37) ~[postgresql-42.2.5.jar:42.2.5]