用用python简单实现简单实现mysql数据同步到数据同步到ElasticSearch的教程的教程
之前博客有用logstash-input-jdbc同步mysql数据到ElasticSearch，但是由于同步时间最少是一分钟一次，无法满足线上业
务，所以只能自己实现一个，但是时间比较紧，所以简单实现一个
思路：思路：
网上有很多思路用什么mysql的binlog功能什么的，但是我对mysql了解实在有限，所以用一个很呆板的办法查询mysql得到数
据，再插入es，因为数据量不大，而且10秒间隔同步一次，效率还可以，为了避免服务器之间的时间差和mysql更新和查询产
生的时间差，所以在查询更新时间条件时是和上一次同步开始时间比较，这样不管数据多少，更新耗时多少都不会少数据，因
为原则是同步不漏掉任何数据，也可以程序多开将时间差和间隔时间差异化，因为用mysql中一个id当作es中的id，也避免了
重复数据
使用：使用：
只需要按照escongif.py写配置文件，然后写sql文件，最后直接执行mstes.py就可以了，我这个也是参考logstash-input-jdbc的
配置形式
MsToEs
|—-esconfig.py（配置文件）
|—-mstes.py（同步程序）
|—-sql_manage.py（数据库管理）
|—-aa.sql（需要用到sql文件）
|—-bb.sql（需要用到sql文件）
sql_manage.py:
# -*-coding:utf-8 -*-
__author__ = "ZJL"
from sqlalchemy.pool import QueuePool
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
import traceback
import esconfig
# 用于不需要回滚和提交的操作
def find(func):
def wrapper(self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except Exception as e:
print(traceback.format_exc())
print(str(e))
return traceback.format_exc()
finally:
self.session.close()
return wrapper
class MysqlManager(object):
def __init__(self):
mysql_connection_string = esconfig.mysql.get("mysql_connection_string")
self.engine = create_engine('mysql+pymysql://'+mysql_connection_string+'?charset=utf8', poolclass=QueuePool,
pool_recycle=3600)
# self.DB_Session = sessionmaker(bind=self.engine)
# self.session = self.DB_Session()
self.DB_Session = sessionmaker(bind=self.engine, autocommit=False, autoflush=True, expire_on_commit=False)
self.db = scoped_session(self.DB_Session)
self.session = self.db()
@find
def select_all_dict(self, sql, keys):
a = self.session.execute(sql)
a = a.fetchall()
lists = [] for i in a:
if len(keys) == len(i):
data_dict = {}
for k, v in zip(keys, i):
data_dict[k] = v
lists.append(data_dict)
else:
return False
return lists
# 关闭
下载后可阅读完整内容，剩余3页未读，立即下载
weixin_38722588
- 粉丝: 6
- 资源: 839
我的内容管理
展开
最新资源
- 十种常见电感线圈电感量计算公式详解
- 军用车辆：CAN总线的集成与优势
- CAN总线在汽车智能换档系统中的作用与实现
- CAN总线数据超载问题及解决策略
- 汽车车身系统CAN总线设计与应用
- SAP企业需求深度剖析：财务会计与供应链的关键流程与改进策略
- CAN总线在发动机电控系统中的通信设计实践
- Spring与iBATIS整合：快速开发与比较分析
- CAN总线驱动的整车管理系统硬件设计详解
- CAN总线通讯智能节点设计与实现
- DSP实现电动汽车CAN总线通讯技术
- CAN协议网关设计：自动位速率检测与互连
- Xcode免证书调试iPad程序开发指南
- 分布式数据库查询优化算法探讨
- Win7安装VC++6.0完全指南：解决兼容性与Office冲突
- MFC实现学生信息管理系统：登录与数据库操作
资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！
点击此处反馈
