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

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: 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>
相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
存储 JSON NoSQL
Python连接MongoDB操作
Python连接MongoDB操作
425 1
|
7月前
|
JSON NoSQL 关系型数据库
Logstash同步MySql数据到Elasticsearch
Logstash同步MySql数据到Elasticsearch
119 0
|
7月前
|
canal 存储 搜索推荐
mysql同步数据到elasticsearch方案详解
通俗易懂,深入浅出讲清楚高级开发人员必备基础数据同步技能
|
关系型数据库 MySQL Go
使用go-mysql-elasticsearch同步mysql数据库信息到ElasticSearch
本文介绍如何使用go-mysql-elasticsearch同步mysql数据库信息到ElasticSearch。1.go-mysql-elasticsearch简介go-mysql-elasticsearch是一个将MySQL数据自动同步到Elasticsearch的服务。
4172 0
|
11月前
|
缓存 编解码 监控
MongoDB - 使用Python操作MongoDB
本文介绍 使用Python操作MongoDB。
383 0
|
关系型数据库 MySQL 开发工具
Python:使用 mysqlsmom 模块实时同步MySQL数据到ElasticSearch
Python:使用 mysqlsmom 模块实时同步MySQL数据到ElasticSearch
117 0
|
canal 消息中间件 关系型数据库
MySQL同步数据到Elasticsearch
随着平台的业务日益增多,基于数据库的全文搜索查询速度较慢,已经无法满足需求。所以,决定基于Elasticsearch 做一个全文搜索平台,支持业务相关的搜索需求。那么第一个问题就是:如何从MySQL同步数据到Elasticsearch?
196 0
MySQL同步数据到Elasticsearch
|
存储 JSON NoSQL
Python 操作mongodb库
MongoDB 是一个基于分布式文件存储的数据库。
207 0
Python 操作mongodb库
|
关系型数据库 MySQL 数据库
Python:mysql-connector-python查询不到新增数据
Python:mysql-connector-python查询不到新增数据
125 0
|
关系型数据库 MySQL
logstash同步MySQL数据到ElasticSearch
logstash同步MySQL数据到ElasticSearch
141 0
logstash同步MySQL数据到ElasticSearch

推荐镜像

更多