Elastic MySQL 连接器是 MySQL 数据源的连接器。它可以帮我们把 MySQL 里的数据同步到 Elasticsearch 中去。在今天的文章里,我来详细地描述如何一步一步地实现。
在下面的展示中,我将使用 Elastic Stack 8.8.2 来进行展示。
无缝集成:将 Elasticsearch 连接到 MongoDB
Enterprise:使用 MySQL connector 同步 MySQL 数据到 Elasticsearch
此连接器在 Elastic 版本 8.5.0 及更高版本中作为本机连接器提供。 要将此连接器用作本机连接器,请满足所有本机连接器 (Native Connector)要求。
此连接器也可用作 Python 连接器框架的连接器客户端。 要将此连接器用作连接器客户端,请满足所有连接器客户端要求。
除了上面链接的共享要求之外,此连接器没有其他先决条件。
要将此连接器用作本机连接器,请使用连接器工作流程。 请参阅本机连接器。
要将此连接器用作连接器客户端,请参阅连接器客户端和框架。
在如下的展示中,我将使用连接器客户端来进行使用。
Elasticsearch
我们可参考我之前的文章 “如何在 Linux,MacOS 及 Windows 上进行安装 Elasticsearch” 来安装 Elasticsearch。特别地,我们需要按照 Elastic Stack 8.x 的安装指南来进行安装。
在 Elasticsearch 终端输出中,找到 elastic 用户的密码和 Kibana 的注册令牌。 这些是在 Elasticsearch 第一次启动时打印的。
我们记下这个密码,并在下面的配置中进行使用。同时它也会生成相应的证书文件:
$ pwd/Users/liuxg/elastic/elasticsearch-8.8.2/config/certs$ lshttp.p12 http_ca.crt transport.p12
为了方便下面的配置,我们把 http_ca.crt 证书拷贝到如下的一个目录中:
mkdir -p ~/connectors-python-configcp http_ca.crt ~/connectors-python-config
保存密码、注册令牌和证书路径名。 你将在后面的步骤中需要它们。如果你对这些操作还不是很熟的话,请参考我之前的文章 “Elastic Stack 8.0 安装 - 保护你的 Elastic Stack 现在比以往任何时候都简单”。
安装 Kibana
我们接下来安装 Kibana。我们可以参考我之前的文章 “如何在 Linux,MacOS 及 Windows 上安装 Elastic 栈中的 Kibana” 来进行我们的安装。特别地,我们需要安装 Kibana 8.2 版本。如果你还不清楚如何安装 Kibana 8.2,那么请阅读我之前的文章 “Elastic Stack 8.0 安装 - 保护你的 Elastic Stack 现在比以往任何时候都简单”。在启动 Kibana 之前,我们可以修改 Kibana 的配置文件如下。添加如下的句子到 config/kibana.yml 中去:
config/kibana.yml
enterpriseSearch.host: http://localhost:3002
然后,我们使用如下的命令来启动 Kibana:
bin/kibana
我们在浏览器中输入上面输出的地址然后输入相应的 enrollment token 就可以把 Kibana 启动起来。
Java安装
你需要安装 Java。版本在 Java 8 或者 Java 11。我们可以参考链接来查找需要的 Java 版本。
Enterprise search 安装
我们在地址 Download Elastic Enterprise Search | Elastic 找到我们需要的版本进行下载。并按照页面上相应的指令来进行按照。如果你想针对你以前的版本进行安装的话,请参阅地址 https://www.elastic.co/downloads/past-releases#app-search。
等我们下载完 Enterprise Search 的安装包,我们可以使用如下的命令来进行解压缩:
$ pwd/Users/liuxg/elastic$ lselasticsearch-8.8.2 kibana-8.8.2elasticsearch-8.8.2-darwin-aarch64.tar.gz kibana-8.8.2-darwin-aarch64.tar.gzenterprise-search-8.8.2.tar.gz$ tar xzf enterprise-search-8.8.2.tar.gz $ cd enterprise-search-8.8.2$ lsLICENSE NOTICE.txt README.md bin config lib metricbeat
如上所示,它含有一个叫做 config 的目录。我们在启动 Enterprise Search 之前,必须做一些相应的配置。我们需要修改 config/enterprise-search.yml 文件。在这个文件中添加如下的内容:
config/enterprise-search.yml
allow_es_settings_modification: truesecret_management.encryption_keys: ['q3t6w9z$C&F)J@McQfTjWnZr4u7x!A%D']elasticsearch.username: elasticelasticsearch.password: "JUYrx8L3WOeG6zysQY2D"elasticsearch.host: https://127.0.0.1:9200elasticsearch.ssl.enabled: trueelasticsearch.ssl.certificate_authority: /Users/liuxg/elastic/elasticsearch-8.8.2/config/certs/http_ca.crtkibana.external_url: http://localhost:5601
在上面,请注意 elasticsearch.password 是我们在 Elasticsearch 安装过程中生成的密码。elasticsearch.ssl.certificate_authority 必须根据自己的 Elasticsearch 安装路径中生成的证书进行配置。在上面的配资中,我们还没有配置 secret_management.encryption_keys。我们可以使用上面的配置先运行,然后让系统帮我们生成。在配置上面的密码时,我们需要添加上引号。我发现在密码中含有 * 字符会有错误的信息。我们使用如下的命令来启动:
bin/enterprise-search
在启动的过程中,我们可以看到生成的用户名及密码信息:
username: enterprise_search password: r9kcpyb5x2g9dken
我们记下这个用户名及密码。在启动的过程中,我们还可以看到一个生成的 secret_session_key:
我们也把它拷贝下来,并添加到配置文件中去:
allow_es_settings_modification: truesecret_management.encryption_keys: ['q3t6w9z$C&F)J@McQfTjWnZr4u7x!A%D'] elasticsearch.username: elasticelasticsearch.password: "JUYrx8L3WOeG6zysQY2D"elasticsearch.host: https://127.0.0.1:9200elasticsearch.ssl.enabled: trueelasticsearch.ssl.certificate_authority: /Users/liuxg/elastic/elasticsearch-8.8.2/config/certs/http_ca.crtkibana.external_url: http://localhost:5601secret_session_key: 3a6d8ab8993a9818728eabd6513fd1c448263be6f5497c8d286bc8be05b87edffd95073582e3277f1e8fb8f753a3ab07a5749ce4394a16f69bdc4acb3d2826aefeature_flag.elasticsearch_search_api: true
为了能够使得我们能够在 App Search 中使用 Elasticsearch 搜索,我们必须设置
feature_flag.elasticsearch_search_api: true。 我们再次重新启动 enterprise search:
./bin/enterprise-search
这次启动后,我们再也不会看到任何的配置输出了。这样我们的 enterprise search 就配置好了。
MySQL
对于本教程,你需要一个供 Logstash 读取的源 MySQL 实例。 MySQL Community Downloads 站点的 MySQL Community Server 部分提供了免费版本的 MySQL。我们可以通过如下的命令来登录 MySQL:
mysql -u root -p
在上面,我们使用 root 的密码来进行登录。针对我的情况,密码为 1234。等我们登录后,我们运行如下的命令:
CREATE DATABASE sample_db;USE sample_db;CREATE TABLE person ( person_id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), age INT);CREATE TABLE address ( address_id INT AUTO_INCREMENT PRIMARY KEY, address VARCHAR(255));INSERT INTO person (name, age) VALUES ('Alice', 30);INSERT INTO person (name, age) VALUES ('Bob', 25);INSERT INTO person (name, age) VALUES ('Carol', 35);INSERT INTO address (address) VALUES ('123 Elm St');INSERT INTO address (address) VALUES ('456 Oak St');INSERT INTO address (address) VALUES ('789 Pine St');
在上面,我们创建了数据库 sample_db,也同时创建了两个表格 address 及 person。
步骤一:下载示例配置文件
下载示例配置文件。 你可以手动下载或运行以下命令:
curl https://raw.githubusercontent.com/elastic/connectors-python/main/config.yml --output ~/connectors-python-config/config.yml
我们可以查看文件:
$ pwd/Users/liuxg/connectors-python-config$ lsconfig.yml http_ca.crt
如果你的目录名称不同,或者你想使用不同的配置文件名,请记住更新 --output 参数值。
步骤二:更新自管理连接器的配置文件
使用以下设置更新配置文件以匹配你的环境:
elasticsearch.host
elasticsearch.password
connector_id
service_type
使用 mysql 作为 service_type 值。 不要忘记取消 yaml 文件源部分中 mysql 的注释。
如果你针对 Elasticsearch 和 Kibana 的 Docker 化版本运行连接器服务,你的配置文件将如下所示:
elasticsearch: host: http://host.docker.internal/:9200 username: elastic password: connector_id: service_type: mysqlsources: # UNCOMMENT "mysql" below to enable the MySQL connector #mongodb: connectors.sources.mongo:MongoDataSource #s3: connectors.sources.s3:S3DataSource #dir: connectors.sources.directory:DirectoryDataSource #mysql: connectors.sources.mysql:MySqlDataSource #network_drive: connectors.sources.network_drive:NASDataSource #google_cloud_storage: connectors.sources.google_cloud_storage:GoogleCloudStorageDataSource #azure_blob_storage: connectors.sources.azure_blob_storage:AzureBlobStorageDataSource #postgresql: connectors.sources.postgresql:PostgreSQLDataSource #oracle: connectors.sources.oracle:OracleDataSource #mssql: connectors.sources.mssql:MSSQLDataSource
请注意,你下载的配置文件可能包含更多条目,因此你需要手动复制/更改适用于您的设置。 通常,你只需要更新 elasticsearch.host、elasticsearch.password、connector_id 和 service_type 即可运行连接器服务。
我们来从 Kibana 界面得到这些配置:
~/connectors-python-config/config.yml
elasticsearch: host: https://192.168.0.3:9200 api_key: "OUkyM1E0a0JrWktfLVd2OTRPZkE6TmkxbUNuN3dROGlrT2cwWlNVaEZKQQ==" ca_certs: "/usr/share/certs/http_ca.crt" ssl: true bulk: queue_max_size: 1024 queue_max_mem_size: 25 display_every: 100 chunk_size: 1000 max_concurrency: 5 chunk_max_mem_size: 5 concurrent_downloads: 10 request_timeout: 120 max_wait_duration: 120 initial_backoff_duration: 1 backoff_multiplier: 2 log_level: infoservice: idling: 30 heartbeat: 300 max_errors: 20 max_errors_span: 600 max_concurrent_content_syncs: 1 max_concurrent_access_control_syncs: 1 job_cleanup_interval: 300 log_level: INFOconnector_id: '8423Q4kBkZK_-Wv9z-en'service_type: 'mysql'sources: # mongodb: connectors.sources.mongo:MongoDataSource # s3: connectors.sources.s3:S3DataSource # dir: connectors.sources.directory:DirectoryDataSource mysql: connectors.sources.mysql:MySqlDataSource # network_drive: connectors.sources.network_drive:NASDataSource # google_cloud_storage: connectors.sources.google_cloud_storage:GoogleCloudStorageDataSource # google_drive: connectors.sources.google_drive:GoogleDriveDataSource # azure_blob_storage: connectors.sources.azure_blob_storage:AzureBlobStorageDataSource # postgresql: connectors.sources.postgresql:PostgreSQLDataSource # oracle: connectors.sources.oracle:OracleDataSource # sharepoint_server: connectors.sources.sharepoint_server:SharepointServerDataSource # mssql: connectors.sources.mssql:MSSQLDataSource # jira: connectors.sources.jira:JiraDataSource # confluence: connectors.sources.confluence:ConfluenceDataSource # dropbox: connectors.sources.dropbox:DropboxDataSource # servicenow: connectors.sources.servicenow:ServiceNowDataSource # sharepoint_online: connectors.sources.sharepoint_online:SharepointOnlineDataSource # github: connectors.sources.github:GitHubDataSource
在上面,请注意:
-
host 是 Elasticsearch 的访问地址
-
api_key 是用来访问 Elasticsearch 的 API key。如果你使用用户名和密码组合,这个就不需要了
-
ca_certs 是用来访问 Elasticsearch 的证书。这个是针对 self-managed 的 Elasticsearch 集群而言的
-
sevice_type 必须是 mysql
-
connector_id 是在上面的配置中生成的。用来标识该连接器
步骤三:运行 Docker 镜像
docker run \-v ~/connectors-python-config:/config \--volume="$PWD/http_ca.crt:/usr/share/certs/http_ca.crt:ro" \--network "elastic" \--tty \--rm \docker.elastic.co/enterprise-search/elastic-connectors:8.8.2.0-SNAPSHOT \/app/bin/elastic-ingest \-c /config/config.yml
当运行完上面的命令后,我们再次回到 Kibana 的界面:
接下来我们来配置 MySQL。由于我们的连接器客户端是在 docker 容器里运行的,二我们的 MySQL 只能在 localhost:3306 进行访问。容器里的代码是没有办法访问到外面的 localhost 地址的。为此,我参照之前的文章 “Kibana:创建一个 webhook alert - Elastic Stack 8.2”。运行如下的命令:
bore local 3306 --to bore.pub
这样 MySQL 就可以被一个公网地址 bore.pub:3332 所访问。我们接下来使用这个地址来进行配置:
我们定于每天的 UTC 零点时间来进行同步。当然,我们也可以选择不定时同步。点击 Save:
我们点击上面的 Sync:
为了验证它是否能够正确地同步新的文档,我们在 MySQL 中添加一个新的文档:
我们在 Kibana 中再次手动 Sync:
由于一些原因,在测试中,我发现在最新的 connector 发布中,它含有 Sync rules,而在我运行的版本中是没有的。它需要在最新的发布版中才有,但是 snapshot 的运行中有一个错误。如果你没有看到这个 Sync rules,请使用如下的命令来启动:
POST .elastic-connectors/_update/YOUR_CONNECTOR_ID{ "doc": { "features": { "sync_rules": { "advanced": { "enabled": true }, "basic": { "enabled": true } } } }}
我们可以通过 Sync rule 来同步我们需要的数据,比如:
[ { "tables": [ "person" ], "query": "SELECT * FROM sample_db.person LIMIT 1;" }, { "tables": [ "address" ], "query": "SELECT * FROM sample_db.address LIMIT 1;" }]
这样,当同步的时候,它只会同步 address 及 person 里的一条数据。
同样,我们可以定义如下的 WHERE query:
[ { "tables": ["person"], "query": "SELECT * FROM sample_db.person WHERE sample_db.person.age > 25;" }]
它只会同步年龄大于 25 岁的 person 里的文档。我们甚至可以做 JOIN query:
[ { "tables": ["person", "tables"], "query": "SELECT * FROM sample_db.person INNER JOIN sample_db.address ON sample_db.person.person_id = sample_db.address.address_id;" }]
我们可以通过如下的方法来查找索引:
GET _cat/indices
我们可以通过如下的命令来查看它的文档:
GET search-mysql/_search
在上面,我们使用本机来安装 MySQL。在实际的测试中,我们可以使用 Docker 更为方便地安装 MySQL:
docker run --name mysql_container -p 3306:3306 -e MYSQL_ROOT_PASSWORD=changeme -e MYSQL_USER=elastic -e MYSQL_PASSWORD=changeme -d mysql:latest
授予用户权限:
docker exec -it mysql_container mysql -u root -p
GRANT ALL PRIVILEGES ON sample_db.* TO 'elastic'@'%';FLUSH PRIVILEGES;
创建数据库及表格:
CREATE DATABASE sample_db;USE sample_db;CREATE TABLE person ( person_id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), age INT);CREATE TABLE address ( address_id INT AUTO_INCREMENT PRIMARY KEY, address VARCHAR(255));INSERT INTO person (name, age) VALUES ('Alice', 30);INSERT INTO person (name, age) VALUES ('Bob', 25);INSERT INTO person (name, age) VALUES ('Carol', 35);INSERT INTO address (address) VALUES ('123 Elm St');INSERT INTO address (address) VALUES ('456 Oak St');INSERT INTO address (address) VALUES ('789 Pine St');
在配置的时候,我们可以参考如下的内容来进行配置:
由于 Connector 和 MySQL 都同时运行于 Docker 中,我们可以使用地址 http://host.docker.internal/ 来填写 MySQL 的访问地址。
在 Kibana 中,它也让我们从源码处下载源码并编译运行 Connector。下面我们来展示如何操作。我们首先使用如下的命令来下载所需要版本的源码:
git clone --branch 8.8 https://github.com/elastic/connectors-python
在上面,我们下载 8.8 版本的源码。我们进入到源码的根目录中,并拷贝相应的证书到该目录中:
cd connectors-python$ pwd/Users/liuxg/python/connectors-pythoncp ~/elastic/elasticsearch-8.8.2/config/certs/http_ca.crt .$ lsCHANGELOG.rst elasticsearch_connectors.egg-infoDockerfile http_ca.crtLICENSE includeMANIFEST.in libMakefile logo-enterprise-search.pngNOTICE.txt pyrightconfig.jsonREADME.md pyvenv.cfgREADME.rst requirementsbin scriptsconfig.yml setup.cfgconnectors setup.pydocs
在我们运行代码之前,我们需要修改源码中的如下文件:
connectors/kibana.py
我们接下来使用如下的命令来进行编译:
我们需要配置配置如下的文件:
config.yml
elasticsearch: host: https://192.168.0.3:9200 api_key: "UFpFTVpJa0JnaFo2aVBURUVxczc6TmcxZ0JpeDNTM1dXNXk0ak02d1piZw==" ca_certs: "./http_ca.crt" ssl: true bulk: queue_max_size: 1024 queue_max_mem_size: 25 display_every: 100 chunk_size: 1000 max_concurrency: 5 chunk_max_mem_size: 5 concurrent_downloads: 10 request_timeout: 120 max_wait_duration: 120 initial_backoff_duration: 1 backoff_multiplier: 2 log_level: infoservice: idling: 30 heartbeat: 300 max_errors: 20 max_errors_span: 600 max_concurrent_syncs: 1 job_cleanup_interval: 300 log_level: INFOnative_service_types: - mongodb - mysql - network_drive - s3 - google_cloud_storage - azure_blob_storage - postgresql - oracle - confluence - dir - sharepoint - mssql - jira# Connector client settingsconnector_id: 'PJELZIkBghZ6iPTE9qs_'service_type: 'mysql'sources: # mongodb: connectors.sources.mongo:MongoDataSource # s3: connectors.sources.s3:S3DataSource # dir: connectors.sources.directory:DirectoryDataSource mysql: connectors.sources.mysql:MySqlDataSource # network_drive: connectors.sources.network_drive:NASDataSource # google_cloud_storage: connectors.sources.google_cloud_storage:GoogleCloudStorageDataSource # azure_blob_storage: connectors.sources.azure_blob_storage:AzureBlobStorageDataSource # postgresql: connectors.sources.postgresql:PostgreSQLDataSource # oracle: connectors.sources.oracle:OracleDataSource # sharepoint: connectors.sources.sharepoint:SharepointDataSource # mssql: connectors.sources.mssql:MSSQLDataSource # jira: connectors.sources.jira:JiraDataSource # confluence: connectors.sources.confluence:ConfluenceDataSource
这个文件和之前的那个很相似,只是因为我们不在 docker 里运行,所以 ca_certs 的路径不同。
我们使用如下的命令来进行运行:
make run
我们可以在 Kibana 中进行如下的配置:
我们接下来按照之前的步骤运行即可。
我们可以看到使用源码运行也可以是使得数据进行同步。
在本文中,我们非常详细地描述如何使用 MySQL connector 来同步 MySQL 和 Elasticsearch 的索引。它使用起来非常方便。如果大家对 Logstash 很熟悉的话,请参阅我之前的文章 “Elastic:开发者上手指南” 中的 “数据库数据同步” 章节。我们还可以使用 Pipeline 对数据进行清洗。这个就不做展示了。
来源地址:https://blog.csdn.net/UbuntuTouch/article/details/131658774