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;
目录
相关文章
|
2月前
|
数据采集 Web App开发 数据可视化
Python零基础爬取东方财富网股票行情数据指南
东方财富网数据稳定、反爬宽松,适合爬虫入门。本文详解使用Python抓取股票行情数据,涵盖请求发送、HTML解析、动态加载处理、代理IP切换及数据可视化,助你快速掌握金融数据爬取技能。
1533 1
|
2月前
|
Java 数据挖掘 数据处理
(Pandas)Python做数据处理必选框架之一!(一):介绍Pandas中的两个数据结构;刨析Series:如何访问数据;数据去重、取众数、总和、标准差、方差、平均值等;判断缺失值、获取索引...
Pandas 是一个开源的数据分析和数据处理库,它是基于 Python 编程语言的。 Pandas 提供了易于使用的数据结构和数据分析工具,特别适用于处理结构化数据,如表格型数据(类似于Excel表格)。 Pandas 是数据科学和分析领域中常用的工具之一,它使得用户能够轻松地从各种数据源中导入数据,并对数据进行高效的操作和分析。 Pandas 主要引入了两种新的数据结构:Series 和 DataFrame。
464 0
|
2月前
|
JSON 算法 API
Python采集淘宝商品评论API接口及JSON数据返回全程指南
Python采集淘宝商品评论API接口及JSON数据返回全程指南
|
2月前
|
JSON API 数据安全/隐私保护
Python采集淘宝拍立淘按图搜索API接口及JSON数据返回全流程指南
通过以上流程,可实现淘宝拍立淘按图搜索的完整调用链路,并获取结构化的JSON商品数据,支撑电商比价、智能推荐等业务场景。
|
4月前
|
机器学习/深度学习 新能源 调度
电力系统短期负荷预测(Python代码+数据+详细文章讲解)
电力系统短期负荷预测(Python代码+数据+详细文章讲解)
418 1
|
4月前
|
缓存 API 网络架构
淘宝item_search_similar - 搜索相似的商品API接口,用python返回数据
淘宝联盟开放平台中,可通过“物料优选接口”(taobao.tbk.dg.optimus.material)实现“搜索相似商品”功能。该接口支持根据商品 ID 获取相似推荐商品,并返回商品信息、价格、优惠等数据,适用于商品推荐、比价等场景。本文提供基于 Python 的实现示例,包含接口调用、数据解析及结果展示。使用时需配置淘宝联盟的 appkey、appsecret 和 adzone_id,并注意接口调用频率限制和使用规范。
|
3月前
|
存储 监控 API
Python实战:跨平台电商数据聚合系统的技术实现
本文介绍如何通过标准化API调用协议,实现淘宝、京东、拼多多等电商平台的商品数据自动化采集、清洗与存储。内容涵盖技术架构设计、Python代码示例及高阶应用(如价格监控系统),提供可直接落地的技术方案,帮助开发者解决多平台数据同步难题。
|
3月前
|
存储 JSON 算法
Python集合:高效处理无序唯一数据的利器
Python集合是一种高效的数据结构,具备自动去重、快速成员检测和无序性等特点,适用于数据去重、集合运算和性能优化等场景。本文通过实例详解其用法与技巧。
151 0
|
5月前
|
存储 Web App开发 前端开发
Python + Requests库爬取动态Ajax分页数据
Python + Requests库爬取动态Ajax分页数据
|
5月前
|
JSON API 数据格式
Python采集京东商品评论API接口示例,json数据返回
下面是一个使用Python采集京东商品评论的完整示例,包括API请求、JSON数据解析

推荐镜像

更多