Python怎么自动化把MySQL当天新增数据增量同步到ES-Python教程-PHP中文网

mysql查当天新增数据需用where create_time >= curdate()并确保字段为datetime/timestamp类型、有索引、考虑逻辑删除与时区一致；es写入需以mysql主键生成_id、用bulk批量导入、显式指定@timestamp；同步需持久化checkpoint；es mapping须手动定义date/keyword/text类型。

python怎么自动化把mysql当天新增数据增量同步到es

MySQL怎么查当天新增数据（带时间戳字段） MySQL 里没有内置“当天新增”语义，得靠业务字段判断。最常见的是 `create_time` 或 `updated_at`，类型必须是 `DATETIME` 或 `TIMESTAMP`，否则 `WHERE create_time >= CURDATE()` 会隐式转换失败或索引失效。

实操建议：

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

先确认字段存在且有索引：SHOW INDEX FROM your_table WHERE Column_name = 'create_time';
用 CURDATE() 而不是 NOW()，避免跨日同步时漏掉 00:00–00:59 的记录
如果业务用的是逻辑删除（is_deleted = 0），查询条件必须带上：WHERE create_time >= CURDATE() AND is_deleted = 0
注意时区：MySQL 服务端时区（SELECT @@time_zone;）要和 Python 程序一致，否则 CURDATE() 返回的“今天”可能错一天

Python怎么把MySQL结果批量写入ES（不覆盖、不重复） ES 没有“INSERT IGNORE”或“ON DUPLICATE KEY UPDATE”，增量同步必须靠 `_id` 控制。推荐用 MySQL 主键（如 `id`）拼接成 ES 的 `_id`，比如 `f"mysql_{row['id']}"`，这样同一条记录反复写入只会更新，不会多出副本。

实操建议：

用 elasticsearch.helpers.bulk() 批量写入，别用单条 index()，吞吐差一个数量级
每条文档构造时显式指定 _id：{"_id": f"mysql_{row['id']}", "_source": {...}}
避免用 create()——它要求文档不存在，但增量任务重启后无法保证这点，容易报 409 Conflict
写入前清空 _source 中的 MySQL 时间字段（如 create_time），换成 ES 可排序的 @timestamp 字段，值用 row['create_time'].isoformat()

怎么避免同步中途断掉后重跑丢数据或重复写入核心是把“最后同步位置”持久化。别用内存变量或临时文件，必须落库或写配置文件。最轻量可靠的做法：建一张 `sync_checkpoint` 表，存 `table_name`、`last_sync_time`、`max_id` 三个字段。

实操建议：

每次同步前先查 SELECT last_sync_time FROM sync_checkpoint WHERE table_name = 'user'
查询 MySQL 时用 WHERE create_time > %s（参数化传入上次时间），而不是 >=，防止重复
同步成功后，用 REPLACE INTO 更新 checkpoint：REPLACE INTO sync_checkpoint VALUES ('user', NOW())
如果表没主键或时间不准，改用 max(id) 做位点，但要注意 id 不连续时会漏数据，此时必须配合 WHERE id > %s ORDER BY id LIMIT 1000 分页拉取

ES mapping 怎么设才不踩坑（尤其日期、数字、text 字段） MySQL 的 `VARCHAR` 直接映射到 ES 的 `text` 会导致无法聚合；`DATETIME` 不显式声明 `date` 类型，会被当成 `keyword`，`range` 查询直接失效。

实操建议：

建索引时手动 PUT mapping，别依赖 dynamic mapping：PUT /my_index + body 包含 "properties": {"created_at": {"type": "date", "format": "strict_date_optional_time||epoch_millis"}}
所有需要聚合/排序的字段（如 status、user_id）设为 "type": "keyword"，哪怕 MySQL 是 INT，也转成字符串再进 ES
text 字段加 "index": false 如果只用于脚本计算、不参与搜索
上线前用 GET /my_index/_mapping 核对，重点看 created_at 是不是 date，title 是不是 text + keyword 多字段

同步逻辑本身不难，难在边界：MySQL 时间字段有没有被 ORM 自动转成本地时区、ES bulk 写入失败后有没有重试机制、checkpoint 更新和数据写入是否在同一个事务里（MySQL 支持，ES 不支持，所以必须先写 ES 成功再更新 checkpoint）。这些地方一漏，就是数据对不上。