Navicat不支持直连Elasticsearch写入，导出JSON再调用API仅适用于一次性、小批量、无更新/删除逻辑的简单场景；真实同步应选用Logstash、canal+Kafka或go-mysql-elasticsearch等专业方案。
navicat 本身不支持直连 elasticsearch 写入，所谓“导出为 json 再调用 api”是可行的，但不是推荐路径——它只适合一次性、小批量（几千条以内）、结构简单、无更新/删除逻辑的场景。真实同步中，这种手动拼 json + curl 的方式极易出错且不可维护。
Navicat 导出 JSON 的实际限制
Navicat 的「导出向导」能生成 JSON 文件，但有硬伤：
- 默认导出的是数组格式（
[{...},{...}]），而Elasticsearchbulk API 要求每行一个index操作指令 + 一行文档，即NDJSON格式 - 字段名不会自动转为小驼峰或下划线，和 ES mapping 不匹配时会写入失败或被丢弃
- NULL 值导出为
null字符串而非真正的 JSONnull，bulk 解析时报错illegal_argument_exception - 时间字段（如
DATETIME）导出为字符串，但未带时区信息，ES 可能解析成 UTC 或报错failed to parse field
如果坚持用 Navicat + API，必须重加工 JSON
不能直接把 Navicat 导出的 JSON 丢给 _bulk。你得先转换格式，例如用 Python 快速处理：
import json
with open("navicat_export.json") as f:
data = json.load(f)
<p>for row in data:</p><h1>1. 构造 index 指令行</h1><pre class='brush:php;toolbar:false;'>print(json.dumps({"index": {"_index": "my_index", "_id": str(row.get("id"))}}))
# 2. 构造文档行（注意清理 null、格式化时间等）
doc = {k: v for k, v in row.items() if v is not None}
if "created_at" in doc:
doc["created_at"] = doc["created_at"] + "+00:00" # 补时区
print(json.dumps(doc))然后保存为 bulk.json，再用：
curl -X POST "http://localhost:9200/_bulk" -H "Content-Type: application/x-ndjson" --data-binary "@bulk.json"
MySQL 到 ES 的真实同步，不该卡在 Navicat 上
Navicat 是数据库 GUI 工具，不是同步管道。一旦数据量超过 1 万行、或需要增量、或表有外键/软删除/时间分区，这套流程就崩了。真正可用的路径只有三条：
-
Logstash+jdbc插件：适合定时全量 + 增量（靠sql_last_value），但单点瓶颈明显，吞吐上不去 -
canal+Kafka+canal-adapter：基于 binlog 的实时同步，要求 MySQL 开ROW模式、配server_id，延迟低但组件多 -
go-mysql-elasticsearch：轻量级 Go 工具，启动即用，自动接mysqldump+binlog，适合中小团队快速落地
Navicat 导出 JSON 这一步，只该出现在验证 mapping 或调试单条文档时，别当主力方案。
最容易被忽略的坑：_id 和 timestamp 处理
即使 JSON 格式对了，Elasticsearch 仍可能静默丢数据：
- 没显式指定
_id？ES 会自动生成，导致同一条 MySQL 记录多次导入变成多条文档 - MySQL 里用
INT存时间戳，但 ES mapping 设成了date类型？bulk 会直接跳过整条 -
lowercase_column_names => false在 Logstash 中没设，字段名大小写混用，filter 阶段取不到值
这些细节不会报错，只会让数据“看起来同步了”，实则漏掉关键字段或重复写入。验证前务必查 _cat/indices 和 _count，再随机 GET 几条比对原始 MySQL 行。
