MySQL查当天新增数据需用WHERE create_time >= CURDATE()并确保字段为DATETIME/TIMESTAMP类型、有索引、考虑逻辑删除与时区一致；ES写入需以MySQL主键生成_id、用bulk批量导入、显式指定@timestamp；同步需持久化checkpoint；ES mapping须手动定义date/keyword/text类型。
MySQL怎么查当天新增数据（带时间戳字段）
MySQL 里没有内置“当天新增”语义，得靠业务字段判断。最常见的是 create_time 或 updated_at，类型必须是 DATETIME 或 TIMESTAMP，否则 WHERE create_time >= CURDATE() 会隐式转换失败或索引失效。
实操建议：
立即学习“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']}"，这样同一条记录反复写入只会更新，不会多出副本。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 用
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 三个字段。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 每次同步前先查
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 查询直接失效。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 建索引时手动 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多字段
