Общая структура показана на рисунке ниже.

    ClickHouse Это два сегмента, состоящие из четырех экземпляров, с двумя копиями каждого сегмента, а голосование и координатор используют использование. ClickHouse Входит в комплект keeper компоненты. Шардинг, реплика, хранитель узел、Кластер Zookeeper、Кластер Kafaka、Debezium-Connector-MySQL Размещение заглушки показано под поверхностью.

Конфигурацияхорошийрепликация главный-подчиненныйназад,существоватьглавная библиотекасоздаватьтестовая библиотекаповерхностьи данные：

-- Создайте базу данных
create database test;

-- учреждатьповерхность
create table test.t1 (
  id bigint(20) not null auto_increment,
  remark varchar(32) default null comment 'Ремарк',createtime timestamp not null default current_timestamp comment 'создаватьвремя',  primary key (id));

-- Вставьте три фрагмента тестовых данных
insert into test.t1 (remark) values («Первая строка: строка1»), («Вторая строка: строка2»), («Третья строка: строка3»);
commit;

    существовать node2 Выполните следующие шаги.

mkdir $KAFKA_HOME/plugins

cd ~
# debezium-connector-mysql
unzip debezium-debezium-connector-mysql-2.4.2.zip -d $KAFKA_HOME/plugins/

# Сначала сделайте резервную копию
cp $KAFKA_HOME/config/connect-distributed.properties $KAFKA_HOME/config/connect-distributed.properties.bak
# редактировать connect-distributed.properties документ
vim $KAFKA_HOME/config/connect-distributed.properties

    Содержание следующее:

bootstrap.servers=node2:9092,node3:9092,node4:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
offset.storage.partitions=3
config.storage.topic=connect-configs
config.storage.replication.factor=3
status.storage.topic=connect-status
status.storage.replication.factor=3
status.storage.partitions=3
offset.flush.interval.ms=10000
plugin.path=/root/kafka_2.13-3.7.0/plugins

scp $KAFKA_HOME/config/connect-distributed.properties node3:$KAFKA_HOME/config/
scp $KAFKA_HOME/config/connect-distributed.properties node4:$KAFKA_HOME/config/
scp -r $KAFKA_HOME/plugins node3:$KAFKA_HOME/
scp -r $KAFKA_HOME/plugins node4:$KAFKA_HOME/

connect-distributed.sh -daemon $KAFKA_HOME/config/connect-distributed.properties 
# Проверьте, есть ли журнал ERROR
grep ERROR ~/kafka_2.13-3.7.0/logs/connectDistributed.out

    Проверятьразъемплагин：

curl -X GET http://node2:8083/connector-plugins | jq

    от Вы можете увидеть приезжать, Кафка на выходе connect Уже признано приезжать MySqlConnector source плагин：

[root@vvml-yz-hbase-test~]#curl -X GET http://node2:8083/connector-plugins | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   403  100   403    0     0   3820      0 --:--:-- --:--:-- --:--:--  3838
[
  {
    "class": "io.debezium.connector.mysql.MySqlConnector",
    "type": "source",
    "version": "2.4.2.Final"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "3.7.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "3.7.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "3.7.0"
  }
]
[root@vvml-yz-hbase-test~]#

    Проверять topic：

kafka-topics.sh --list --bootstrap-server node2:9092,node3:9092,node4:9092

    от Вы можете увидеть приезжать, Кафка на выходе connect Автоматически создавать при запуске connect-configs、connect-offsets、connect-status три topic：

[root@vvml-yz-hbase-test~]#kafka-topics.sh --list --bootstrap-server node2:9092,node3:9092,node4:9092
__consumer_offsets
connect-configs
connect-offsets
connect-status
[root@vvml-yz-hbase-test~]#

    Debezium это хорошо известное средство для чтения и синтаксического анализа. MySQL Binlog инструменты. это будет KafkaConnect Интегрируйте в качестве соединителя и Kafka Тема меняется каждый раз.

    По умолчанию Дебезиум Пойду в Kafka Выдавать предварительное и пост-состояние каждой записи для каждого действия, что сложно ClickHouse Kafka поверхностьанализировать。также,существоватьосуществлятьудалитьдействоватьслучай（Clickhouse Также не может быть разобран), он создаст tombstone записать, то есть иметь Null запись стоимости. Следующая поверхность демонстрирует такое поведение.

    существовать Debezium Конфигурацияпосланникиспользовать ExtractNewRecod конвертер для решения этой проблемы. Благодаря этому варианту Дебезиум Зарезервировано только для создателя/Обновлениядействовать after статус и игнорировать before состояние. Но недостатком является то, что он удаляет Delete запись и запись надгробия, другими словами, больше не захватывать и не удалять действия. Далее объясните, как решить эту проблему.

"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"

    Чтобы отследить удаление действия, необходимо добавить переопределение Конфигурация, как показано ниже:

"transforms.unwrap.delete.handling.mode":"rewrite"

    Debezium Makeusethis Конфигурация добавить поля \_\_deleted, для delete действоватьдля правда, для других действий ЛОЖЬ. Таким образом, удаление будет содержать предыдущее состояние, а также \_\_deleted:true поле.

    существовать Предоставьте вышеуказанное Конфигурацияслучай,Обновить запись（Каждый столбец, кроме первичного ключа）выпустит новыйсостояниепростоты Записывать.в целомсуществоватьв системе реляционных баз данных,Обновленные записи заменят предыдущую запись.,носуществовать ClickHouse нет. Из соображений производительности ClickHouse Воляобновление уровня строкидля Вставка нескольких версий。существовать В этом примере,MySQL в test.t1 поверхностьк id Столбец как первичный ключ, если он обновлен. remark Список,существовать ClikHouse , вы в конечном итоге получите дублирующиеся записи, что означает id То же самое, но remark другой!

    К счастью, есть способы справиться с этой ситуацией. По умолчанию Дебезиум Будет одна запись удаления и одна запись, используемая для обновления первичного ключа. Итак, если источник обновлен id, он выдаст идентификатор с предыдущим id удаленных записей и одна с новыми id создать запись. с \_\_deleted=ture Предыдущая запись для поля будет заменена. CH в stall Записывать.Ранназад,Можетсуществовать Фильтрация подсказок в представленияхудалитьиз Записывать.Можетделатьиспользоватьк下选项Воля Эта поездкадля Расширятьприезжатьдругой Список：

"message.key.columns": "test.t1:id;test.t1:remark;test.t1:createtime"

     нужно изменить ключевой столбец коннектора Дебезиум Воля Эти Списокиспользоватькак первичный ключ,Вместо первичного ключа исходной поверхности по умолчанию. поэтому,ибаза данныхизсвязанный с записьюиздругойдействоватьможет со временем появитьсясуществовать Kafka в Другие разделы。из-за записисуществовать Потеря порядка в разных разделах,если не обеспечено ClickHouse Последовательная ключевая сумма Debezium ключи сообщений одинаковы, в противном случае это может привести к Clikchouse в Данные непоследовательный.

    сейчассуществовать,проходить Совместите все вышеперечисленные варианты с обычнымииспользоватьопцион путсуществовать Вместе,Будет иметь полнофункциональную конфигурацию Debezium., способная обрабатывать ClickHouse Требуются любые изменения.

# редактироватьдокумент
vim $KAFKA_HOME/plugins/source-mysql.json

    Содержание следующее:

{
 "name": "mysql-source-connector",
 "config": {
     "connector.class": "io.debezium.connector.mysql.MySqlConnector",
     "database.hostname": "172.18.16.156",
     "database.port": "3307",
     "database.user": "dba",
     "database.password": "123456",
     "database.server.id": "1563307",
     "database.server.name": "dbserver1",
     "database.include.list": "test",
     "table.include.list": "test.t1",
     "topic.prefix": "mysql-clickhouse-test",
     "schema.history.internal.kafka.bootstrap.servers": "node2:9092,node3:9092,node4:9092",
     "schema.history.internal.kafka.topic": "schemahistory.mysql-clickhouse-test",
     "message.key.columns": "test.t1:id;test.t1:remark;test.t1:createtime",
     "transforms":"unwrap",
     "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
     "transforms.unwrap.delete.handling.mode": "rewrite"
     }
 }

# создавать connector
curl -X POST -H 'Content-Type: application/json' -i 'http://node2:8083/connectors' -d @"/root/kafka_2.13-3.7.0/plugins/source-mysql.json"; echo
# Проверять connector состояние
curl -X GET http://node2:8083/connectors/mysql-source-connector/status | jq
# Проверять topic
kafka-topics.sh --list --bootstrap-server node2:9092,node3:9092,node4:9092

    от Вы можете увидеть приезжать, mysql-source-connector в выводе состояниедля RUNNING,И автоматически создаётся три topic：

[root@vvml-yz-hbase-test~]#curl -X POST -H 'Content-Type: application/json' -i 'http://node2:8083/connectors' -d @"/root/kafka_2.13-3.7.0/plugins/source-mysql.json"; echo
HTTP/1.1 201 Created
Date: Thu, 25 Apr 2024 03:47:26 GMT
Location: http://node2:8083/connectors/mysql-source-connector
Content-Type: application/json
Content-Length: 818
Server: Jetty(9.4.53.v20231009)

{"name":"mysql-source-connector","config":{"connector.class":"io.debezium.connector.mysql.MySqlConnector","database.hostname":"172.18.16.156","database.port":"3307","database.user":"dba","database.password":"123456","database.server.id":"1563307","database.server.name":"dbserver1","database.include.list":"test","table.include.list":"test.t1","topic.prefix":"mysql-clickhouse-test","schema.history.internal.kafka.bootstrap.servers":"node2:9092,node3:9092,node4:9092","schema.history.internal.kafka.topic":"schemahistory.mysql-clickhouse-test","message.key.columns":"test.t1:id;test.t1:remark;test.t1:createtime","transforms":"unwrap","transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState","transforms.unwrap.delete.handling.mode":"rewrite","name":"mysql-source-connector"},"tasks":[],"type":"source"}
[root@vvml-yz-hbase-test~]#curl -X GET http://node2:8083/connectors/mysql-source-connector/status | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   182  100   182    0     0  24045      0 --:--:-- --:--:-- --:--:-- 26000
{
  "name": "mysql-source-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "172.18.4.188:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "172.18.4.188:8083"
    }
  ],
  "type": "source"
}
[root@vvml-yz-hbase-test~]#kafka-topics.sh --list --bootstrap-server node2:9092,node3:9092,node4:9092
__consumer_offsets
connect-configs
connect-offsets
connect-status
mysql-clickhouse-test
mysql-clickhouse-test.test.t1
schemahistory.mysql-clickhouse-test
[root@vvml-yz-hbase-test~]#

    ClickHouse Можетвыгодаиспользовать Kafka настольный движок Воля Kafka Пластинки помещены в поверхность. Необходимо определить три объекта: Kafka поверхность, основная поверхность и материализованный вид потребителя.

create database db2 on cluster cluster_2S_2R;

CREATE TABLE db2.kafka_t1 on cluster cluster_2S_2R
(
    `id` Int64,
    `remark` Nullable(String),
    `createtime` String,
    `__deleted` String
)
ENGINE = Kafka('node2:9092,node3:9092,node4:9092', 'mysql-clickhouse-test.test.t1', 'clickhouse', 'JSONEachRow');

    Основная поверхность имеет исходную структуру и \_\_deleted поле. Слово «использовать» здесь ReplicationReplacingMergeTree, поскольку удаленные или обновленные записи необходимо заменить. stall Записывать.

-- создаватьместныйповерхность
CREATE TABLE db2.stream_t1 on cluster cluster_2S_2R
(
    `id` Int64,
    `remark` Nullable(String),
    `createtime` timestamp,
    `__deleted` String
)
ENGINE = ReplicatedReplacingMergeTree(
    '/clickhouse/tables/{shard}/db2/t1',
    '{replica}'
)
ORDER BY (id, createtime)
SETTINGS index_granularity = 8192;

-- создать распределенную поверхность с первичным ключом исходной поверхности id Как ключ шардинга, он гарантированно будет таким же id изпадение данныхсуществовать На том же осколке
create table db2.t1_replica_all on cluster cluster_2S_2R
as db2.stream_t1
engine = Distributed(cluster_2S_2R, db2, stream_t1, id);

    существоватьсоздаватьматериализованный видвперед,Сначала остановите библиотеку MySQLот копировать. от библиотеки стоп копировать,Не влияет на нормальное использование основной библиотеки.,На бизнес это не повлияет. в это времяот Библиотекаиз Данные находятся в состоянии покоясостояние,без изменений,Это упрощает получение существующих данных.。Ранназадсоздаватьматериализованный вид时会自动Воля Запись данных db2.t1\_replica\_all переписыватьсяизместныйповерхностьсередина。Изназадсуществовать ClickHouse кластервв любом случае,Все могутотматериализованный видсередина Запросприезжатьпоследовательныйиз MySQL Данные о запасах.

-- MySQL отбиблиотека остановленакопировать
stop slave;

    Kafka Каждая запись поверхности считывается только один раз, поскольку ее потребительская группа изменит смещение и не может быть прочитана дважды. Поэтому необходимо определить основную поверхность и приложить к каждой материализованный вид. Kafka поверхностьматериализация записиприезжатьэто：

-- Обратите внимание на обработку временных меток
CREATE MATERIALIZED VIEW db2.consumer_t1 on cluster cluster_2S_2R
TO db2.t1_replica_all
(
    `id` Int64,
    `remark` Nullable(String),
    `createtime` timestamp,
    `__deleted` String
) AS
SELECT id, remark, addHours(toDateTime(substring(createtime,1,length(createtime)-1)),8) createtime, __deleted FROM db2.kafka_t1;

Наконец, каждую удаленную запись необходимо фильтровать.,и иметь последние записи,В случае, если разные записи имеют один и тот же ключ сортировки. Для этого неявно можно определить простое представление:

CREATE VIEW db2.t1 on cluster cluster_2S_2R
(
    `id` Int64,
    `remark` Nullable(String),
    `createtime` String,
    `__deleted` String
) AS
SELECT *
FROM db2.consumer_t1
FINAL
WHERE __deleted = 'false';

    от clickhouse Просмотрите данные инвентаризации запроса:

vvml-yz-hbase-test.172.18.4.126 :) select * from db2.t1;

SELECT *
FROM db2.t1

Query id: 2a51fd5e-6b4f-4b78-b522-62b7be32535b

┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  2 │ Вторая строка: строка2 │ 2024-04-25 11:51:07 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘
┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  1 │ Первая строка: строка1 │ 2024-04-25 11:51:07 │ false     │
│  3 │ Третий ряд: ряд3 │ 2024-04-25 11:51:07 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘

3 rows in set. Elapsed: 0.007 sec. 

vvml-yz-hbase-test.172.18.4.126 :) 

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

-- MySQL Изменить данные в основной базе данных
insert into test.t1 (remark) values («Четвертая строка: строка4»);
update test.t1 set remark = 'Пятая строка: row5' where id = 4;
delete from test.t1 where id =1;
insert into test.t1 (remark) values («Шестая строка: строка6»);
 
-- MySQL от стартапа библиотеки копировать
начинать slave;

    в это время MySQL Данные следующие:

mysql> select * from test.t1;
+----+------------------+---------------------+
| id | remark           | createtime          |
+----+------------------+---------------------+
|  2 | Вторая строка: строка2     | 2024-04-25 11:51:07 |
|  3 | Третий ряд: ряд3     | 2024-04-25 11:51:07 |
|  4 | Пятый ряд: row5     | 2024-04-25 11:56:29 |
|  5 | Ряд 6: ряд6     | 2024-04-25 11:56:29 |
+----+------------------+---------------------+
4 rows in set (0.00 sec)

    от clickhouse Просмотр запроса для дополнительных данных:

vvml-yz-hbase-test.172.18.4.126 :) select * from db2.t1;

SELECT *
FROM db2.t1

Query id: b34bb37b-091b-490e-b55b-a0e9eedf5573

┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  2 │ Вторая строка: строка2 │ 2024-04-25 11:51:07 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘
┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  4 │ Пятый ряд: row5 │ 2024-04-25 11:56:29 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘
┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  3 │ Третий ряд: ряд3 │ 2024-04-25 11:51:07 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘
┌─id─┬─remark───────┬─createtime──────────┬─__deleted─┐
│  5 │ Ряд 6: ряд6 │ 2024-04-25 11:56:29 │ false     │
└────┴──────────────┴─────────────────────┴───────────┘

4 rows in set. Elapsed: 0.008 sec. 

vvml-yz-hbase-test.172.18.4.126 :) 

    Вы можете увидеть, как добраться, дополнительные данные были MySQL синхронный,сейчассуществоватьот ClickHouse Просмотр данных запроса и MySQL последовательный.

    Проверять Kafka Потребление:

kafka-consumer-groups.sh --bootstrap-server node2:9092,node3:9092,node4:9092 --describe --group clickhouse

    Вывод следующий:

[root@vvml-yz-hbase-test~]#kafka-consumer-groups.sh --bootstrap-server node2:9092,node3:9092,node4:9092 --describe --group clickhouse

GROUP           TOPIC                         PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                                                  HOST            CLIENT-ID
clickhouse      mysql-clickhouse-test.test.t1 0          8               8               0               ClickHouse-vvml-yz-hbase-test.172.18.4.126-db2-kafka_t1-26e6aa8e-1f08-4491-8af7-f1822f1a7e94 /172.18.4.126   ClickHouse-vvml-yz-hbase-test.172.18.4.126-db2-kafka_t1
[root@vvml-yz-hbase-test~]#

    Вы можете увидеть приезжать, смещение последнего использованного сообщения - 8, MySQL Запасы и дополнительные данные обновлены. Kafka Сообщение синхронизировано приезжать ClickHouse。