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

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS PostgreSQL,高可用系列 2核4GB
RDS MySQL Serverless 高可用系列,价值2615元额度,1个月
简介: 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;
目录
相关文章
|
人工智能 Java API
Google Gemini API 接口调用方法
Google 最近发布的 Gemini 1.0 AI 模型通过其升级版,Gemini,标志着公司迄今为止最为强大和多功能的人工智能技术的突破。
|
存储 安全 Linux
|
9月前
|
负载均衡 网络协议 网络性能优化
动态IP代理技术详解及网络性能优化
动态IP代理技术通过灵活更换IP地址,广泛应用于数据采集、网络安全测试等领域。本文详细解析其工作原理,涵盖HTTP、SOCKS代理及代理池的实现方法,并提供代码示例。同时探讨配置动态代理IP后如何通过智能调度、负载均衡、优化协议选择等方式提升网络性能,确保高效稳定的网络访问。
928 2
|
8月前
|
机器学习/深度学习 资源调度 算法
机器学习领域必知数学符号与概念(一)
本文介绍了一些数学符号以及这些符号的含义。
635 65
|
存储 缓存 网络协议
网络丢包排查方法
网络丢包排查方法
|
12月前
|
算法 Java 数据安全/隐私保护
国密加密算法简介
国密指国家密码局认定的国产密码算法,主要包括SM1、SM2、SM3、SM4等,并持续完善。SM1是对称加密算法,加密强度与AES相当,需加密芯片支持;SM2是非对称加密,基于ECC算法,签名和密钥生成速度优于RSA;SM3为杂凑算法,安全性高于MD5;SM4为对称加密算法,用于无线局域网标准。本文提供使用Java和SpringBoot实现SM2和SM4加密的示例代码及依赖配置。更多国密算法标准可参考国家密码局官网。
1663 2
|
12月前
|
SQL Shell 数据库连接
死磕xxl-job(二)
死磕xxl-job(二)
|
NoSQL MongoDB 关系型数据库
13个Mongodb GUI可视化管理工具,总有一款适合你
本文介绍了13个好用的MongoDB可视化工具。Robomongo,MongoDB Compass,phpMoAdmin等
111525 0
13个Mongodb GUI可视化管理工具,总有一款适合你
|
存储 人工智能 自然语言处理
AI时代你一定要知道的Agent概念
这两年,随着人工智能(AI)和计算能力的发展,AI应用的落地速度大大加快。以ChatGPT为代表的AI应用迅速火遍全球,成为打工人的常用工具。紧接着,多模态、AI Agent等各种高大尚的名词也逐渐进入大众视野,吸引了大量关注。那么,到底什么是AI Agent?下文半支烟将带你详细了解这个概念。
|
JSON JavaScript 数据格式
【深入探究C++ JSON库】解析JSON元素的层级管理与遍历手段
【深入探究C++ JSON库】解析JSON元素的层级管理与遍历手段
1884 2