Python:使用 mysqlsmom 模块实时同步MySQL数据到ElasticSearch

简介: Python:使用 mysqlsmom 模块实时同步MySQL数据到ElasticSearch

mysqlsmom 文档: https://mysqlsmom.readthedocs.io/en/latest/hello.html

github: https://github.com/m358807551/mysqlsmom


环境要求:

1、python2.7
2、redis
3、Mysql 配置 binlog-format=row

安装

 pip install mysqlsmom

全量同步

# 创建全量同步配置文件


$ mom new test_mom/init_config.py -t init --force

# 编辑配置文件
$ vim ./test_mom/init_config.py # 按注释提示修改配置

# 开始同步
$ mom run -c ./test_mom/init_config.py

增量同步

配置三个文件

test_mom
├── binlog_config.py # 配置文件
├── my_filters.py # 过滤器 配置于 watched
└── my_handlers.py # 处理器 配置于 pipeline

新建配置

mom new test_mom/binlog_config.py -t binlog --force

1、binlog_config.py

# coding=utf-8

STREAM = "BINLOG" # "BINLOG" or "INIT"
SERVER_ID = 99
SLAVE_UUID = name

# 一次同步 BULK_SIZE 条数据到elasticsearch,不设置该配置项默认为1
BULK_SIZE = 1

BINLOG_CONNECTION = {
'host': '127.0.0.1',
'port': 3306,
'user': 'root',
'passwd': '123456'
}

# redis存储上次同步位置等信息
REDIS = {
"host": "127.0.0.1",
"port": 6379,
"db": 0,
# "password": "password", # 不需要密码则注释或删掉该行
}

NODES = [{"host": "127.0.0.1", "port": 9200}]

TASKS = [
{
"stream": {
"database": "demo",
"table": "student"
},
"jobs": [{
"actions": ["insert", "update"],
"watched": {
"filter_display": {}
},
"pipeline": [
{"only_fields": {"fields": ["id", "name", "age"]}},
{"change_name": {"key": "name", "prefix": "hot-"}},
{"set_id": {"field": "id"}}
],
"dest": {
"es": {
"action": "upsert",
"index": "demo",
"type": "student",
"nodes": NODES
}
}
},
{
"actions": ["delete"],
"pipeline": [
# {"only_fields": {"fields": ["id", "name", "age"]}},
{"set_id": {"field": "id"}}
],
"dest": {
"es": {
"action": "delete",
"index": "demo",
"type": "student",
"nodes": NODES
}
}
}
]
}
]

CUSTOM_ROW_HANDLERS = "./my_handlers.py"
CUSTOM_ROW_FILTERS = "./my_filters.py"

自定义处理器 my_handlers.py

# -- coding: utf-8 --


import copy

def change_name(row, key, prefix):
new_row = copy.deepcopy(row)
new_row[key] = "{}{}".format(prefix, row[key])
# 返回数据字典,下一工序继续处理
return new_row

自定义过滤器 my_filters.py

# -- coding: utf-8 --

def filter_display(event):
# 返回True 或 False,使用或丢弃
return event"values" == 1

启动

mom run -c test_mom/binlog_config.py 


            </div>
相关实践学习
以电商场景为例搭建AI语义搜索应用
本实验旨在通过阿里云Elasticsearch结合阿里云搜索开发工作台AI模型服务,构建一个高效、精准的语义搜索系统,模拟电商场景,深入理解AI搜索技术原理并掌握其实现过程。
ElasticSearch 最新快速入门教程
本课程由千锋教育提供。全文搜索的需求非常大。而开源的解决办法Elasricsearch(Elastic)就是一个非常好的工具。目前是全文搜索引擎的首选。本系列教程由浅入深讲解了在CentOS7系统下如何搭建ElasticSearch,如何使用Kibana实现各种方式的搜索并详细分析了搜索的原理,最后讲解了在Java应用中如何集成ElasticSearch并实现搜索。 &nbsp;
目录
相关文章
|
存储 算法 BI
经典加解密算法——香农算法介绍(附源码!)
经典加解密算法——香农算法介绍(附源码!)
536 0
经典加解密算法——香农算法介绍(附源码!)
|
关系型数据库 MySQL Docker
6. 修改docker端口 (映射宿主机和docker容器中的端口)
6. 修改docker端口 (映射宿主机和docker容器中的端口)
3595 0
6. 修改docker端口 (映射宿主机和docker容器中的端口)
|
定位技术
Echarts实战案例代码(10):echarts结合世界所有国家地图数据集geojson的(英文翻译映射)解决方案
Echarts实战案例代码(10):echarts结合世界所有国家地图数据集geojson的(英文翻译映射)解决方案
1331 0
Echarts实战案例代码(10):echarts结合世界所有国家地图数据集geojson的(英文翻译映射)解决方案
|
存储 缓存 测试技术
现代化实时数仓 SelectDB 再次登顶 ClickBench 全球数据库分析性能排行榜!
近日,在 ClickHouse 发起的分析型数据库性能测试排行榜 ClickBench(https://benchmark.clickhouse.com/)中,现代化实时数仓 SelectDB 时隔两年后再次登顶,在全部近百款数据库和数十种机型中,性能表现位居总榜第一!
590 1
|
前端开发 Java 应用服务中间件
【异常解决】java程序连接MinIO报错The request signature we calculated does not match the signature you provided.
【异常解决】java程序连接MinIO报错The request signature we calculated does not match the signature you provided.
1364 0
|
存储 JSON 前端开发
【Spring MVC学习笔记 五】SpringMVC框架整合Jackson工具
【Spring MVC学习笔记 五】SpringMVC框架整合Jackson工具
868 0
|
监控
zookeeper集群安装和简单命令学习
最近在学习zookeeper的使用,所以自己在电脑上安装了个简单版的zookeeper集群,在这里将整个安装过程和一些简单的zookeeper命令写下来,后续再随着自己的不断学习中会抽空完善 首先你需要准备三台虚拟机,我这边创建了三台虚拟机,ip分别为192.
1242 0
|
3天前
|
数据采集 人工智能 安全
|
12天前
|
云安全 监控 安全