Debezium 中的连接：MySQL 到 Elasticsearch

MySQL

使用Debezium将MySQL连接到Elasticsearch

将MySQL数据库中的数据实时同步到Elasticsearch是许多企业中常见的需求。Debezium是一个开源的分布式平台，可以捕获数据库的变化并将其转发到不同的目标系统。在本文中，我们将探讨如何使用Debezium来连接MySQL数据库与Elasticsearch，并实现实时的数据同步。

准备工作

在开始之前，我们需要准备以下工具和环境：

1. MySQL数据库：确保已安装并运行MySQL数据库。

2. Elasticsearch：确保已安装并运行Elasticsearch。

3. Kafka：Debezium使用Kafka作为消息队列，因此需要安装并运行Kafka。

安装Debezium插件

首先，我们需要安装Debezium的MySQL插件。可以从Debezium官方网站下载最新版本的插件，并将其解压到指定的目录中。然后，在MySQL的配置文件中添加以下行：

plugin-load = "MySQLx=usr/lib/MySQLx/plugin/MySQLx.so"

保存并重新启动MySQL服务器，以使插件生效。

配置Kafka连接

接下来，我们需要配置Debezium连接到Kafka。在Debezium的安装目录中，找到并编辑debezium-connector-MySQL.properties文件。在文件中添加以下配置：

name=MySQL-connector
Database.hostname=localhost
Database.port=3306
Database.user=your_MySQL_username
Database.password=your_MySQL_password
Database.server.id=1
Database.server.name=my-app-connector
Database.whitelist=your_Database_name

确保将your_MySQL_username和your_MySQL_password替换为实际的MySQL用户名和密码，并将your_Database_name替换为要同步的数据库名称。

配置Elasticsearch连接

然后，我们需要配置Debezium连接到Elasticsearch。在同一个文件中，添加以下配置：

Database.history.kafka.bootstrap.servers=localhost:9092
Database.history.kafka.topic=dbhistory.my-app-connector

这些配置将告诉Debezium将数据库变化的历史记录写入到Kafka的指定主题中。

启动Debezium

现在，我们已经完成了所有必要的配置。启动Kafka和Elasticsearch，并在Debezium的安装目录中运行以下命令启动Debezium：

./bin/connect-standalone.sh ./config/worker.properties ./config/my-app-connector.properties

这将启动Debezium并开始监听MySQL数据库的变化。

案例代码

下面是一个简单的Java代码示例，演示如何使用Debezium将MySQL连接到Elasticsearch：

Java
import io.debezium.config.Configuration;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.embedded.EmbeddedEngine.Builder;
import org.apache.kafka.connect.JSon.JSonDeserializer;
import org.apache.kafka.connect.JSon.JSonSerializer;
import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
import org.apache.kafka.connect.storage.OffsetBackingStore;
import Java.util.Properties;
public class MySQLToElasticsearchConnector {
    public static void mAIn(String[] args) {
        Properties props = new Properties();
        props.setProperty("name", "MySQL-connector");
        props.setProperty("connector.class", "io.debezium.connector.MySQL.MySQLConnector");
        props.setProperty("Database.hostname", "localhost");
        props.setProperty("Database.port", "3306");
        props.setProperty("Database.user", "your_MySQL_username");
        props.setProperty("Database.password", "your_MySQL_password");
        props.setProperty("Database.server.id", "1");
        props.setProperty("Database.server.name", "my-app-connector");
        props.setProperty("Database.whitelist", "your_Database_name");
        Configuration config = Configuration.from(props);
        Builder builder = EmbeddedEngine.create()
                .using(config)
                .using(deserializer)
                .using(serializer)
                .using(offsetBackingStore);
        EmbeddedEngine engine = builder.build();
        engine.run();
    }
}

这个例子展示了如何配置Debezium连接到MySQL数据库，并使用自定义的反序列化器和序列化器来处理数据。你可以根据实际需求进行修改和扩展。

通过使用Debezium，我们可以轻松地将MySQL数据库连接到Elasticsearch，并实现实时的数据同步。本文介绍了如何安装和配置Debezium插件，并提供了一个简单的Java代码示例。希望这对于需要将MySQL数据传输到Elasticsearch的开发人员和数据工程师们有所帮助。