个别场景下，开发提需求，需要把某个MySQL里面指定数据同步到ES中，希望能有一个通用的脚本，用于特殊场景下的补数据或者临时性的数据同步。
注意： python es包的版本如果和es服务端的版本不一致的话，可能遇到报错。把python es的包版本换成和server端一致的版本即可。
下面的这个脚本，是用python+django+celery来实现上述功能的。核心代码如下：
方法1 逐条记录同步
代码运行次数：0
Cloud Studio 代码运行
# -*- coding: utf-8 -*-
# 根据MySQL表的update_time列同步增量数据到ES中，注意必须带上esId这个字段，这个值是作为ES的_id的
import os
import sys
import time
import mysql.connector
from elasticsearch import Elasticsearch
def sync(task_name, mysql_host, mysql_port, mysql_user, mysql_pass, sql_condition, es_addr, es_index_name):
try:
mydb = mysql.connector.connect(
host=mysql_host,
port=mysql_port,
user=mysql_user,
passwd=mysql_pass,
ssl_disabled=True,
)
# 注意这里把结果输出为dict格式
mycursor = mydb.cursor(dictionary=True)
except Exception as e:
# 发送钉钉告警
sys.exit(10)
mycursor.execute(sql_condition)
res = mycursor.fetchall()
es = Elasticsearch(es_addr, timeout=60, max_retries=10, retry_on_timeout=True)
for i in res:
try:
es.index(index=es_index_name, body=dict(i), id=i['esId'])
except Exception as e:
# 发送钉钉告警
pass
finally:
continue
if __name__ == '__main__':
t1 = time.time()
sync("DBA-测试",
"192.168.31.181", 3306, "dts", "dts",
"select id as esId,k,c,pad,ccc from sbtest.sbtest3",
['http://192.168.31.181:8989'],
'dba-test-new-2',
)
t2 = time.time()
print(t2-t1)
方法2 批量同步方式【推荐用这种】
代码运行次数：0
# -*- coding: utf-8 -*-
import time
import os
import sys
from elasticsearch import Elasticsearch,helpers
import mysql.connector
def sync(task_name, mysql_host, mysql_port, mysql_user, mysql_pass, sql_condition, es_addr, es_index_name):
try:
mydb = mysql.connector.connect(
host=mysql_host,
port=mysql_port,
user=mysql_user,
passwd=mysql_pass,
ssl_disabled=True,
)
mycursor = mydb.cursor(dictionary=True)
except Exception as e:
sys.exit(10)
try:
mycursor.execute(sql_condition)
res = mycursor.fetchall()
es = Elasticsearch(es_addr, request_timeout=60, max_retries=10, retry_on_timeout=True)
# 准备bulk操作的数据
actions = (
{
"_index": es_index_name,
"_id": i['esId'],
"_source": dict(i)
} for i in res
)
# 使用helpers.bulk进行批量写入
success, _ = helpers.bulk(es, actions)
print(f"Indexed {success} documents.")
except Exception as e:
print(f"Error indexing documents to Elasticsearch: {e}")
finally:
mycursor.close()
mydb.close()
if __name__ == '__main__':
t1 = time.time()
sync("DBA-测试",
"192.168.31.181", 3306, "dts", "dts",
"select id as esId,k,c,pad,ccc from sbtest.sbtest3",
['http://192.168.31.181:8989'],
'dba-test-new-2',
)
t2 = time.time()
print(t2-t1)
耗时
MySQL端记录数： 94326
行记录demo：
[sbtest]> select * from sbtest3 limit 10,1 \G
*************************** 1. row ***************************
id: 5685
k: 50479
c: 91674238320-66576604523-84892597271-42298112537-31748098687-87592861234-27236853894-78260103493-10155978333-85784381566
pad: 50437784151-86772162187-31166376983-54827989967-72340867827
ccc: NULL
1 row in set (0.00 sec)
逐条提交： 耗时 389秒
批量提交： 耗时 12秒
设置索引mapping，否则可能出现时间列格式等其它问题
代码运行次数：0
Cloud Studio 代码运行
PUT dba-test
{
"mappings": {
"properties": {
"orderNo": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"esId": {
"type": "keyword"
},
"name": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"sex": {
"type": "keyword"
},
"update_time": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
},
"remark": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
}
}
}
celery 任务配置类似如下：
代码运行次数：0
Cloud Studio 代码运行
{
"task_name": "dba-test",
"mysql_host": "192.168.1.11",
"mysql_port": 3306,
"mysql_user": "dts",
"mysql_pass": "dts",
"sql_condition": "select a as esId,a as orderNo,name,sex,update_time,remark from sbtest.t1 where update_time>DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 5 MINUTE), '%Y-%m-%d %H:%i:%s')",
"es_addr": [
"192.168.1.110:9200",
"192.168.1.111:9200",
"192.168.1.112:9200"
],
"es_index_name": "dba-test"
}
根据实际需要，celery里面可以使用周期性任务或者一次性任务。
此外，这里的sql_condition 也支持复杂条件，例如直接进行2表关联取值（注意esId不要重复就行）：
"sql_condition": "select b.a as esId,a.update_time,a.name,a.sex,b.addr,b.job from sbtest.t1 a inner join sbtest.t2 b on a.name=b.name where a.update_time>DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 5 MINUTE), '%Y-%m-%d %H:%i:%s')",
生产上，还需要接钉钉告警，如果数据同步失败，会及时通知，类似如下：
