ES 作为一个分布式搜索引擎，从扩展能力和搜索特性上而言无出其右，然而它有自身的弱势存在，其作为近实时存储系统，由于其分片和复制的设计原理，也使其在数据延迟和一致性方面都是无法和 OLTP（Online Transaction Processing）系统相媲美的。

ES 基础写入原理

ES 直连写入

1. 不易于接入多机房容灾部署，目前 ES 容灾机房都属于独立部署，独立读写模式，所以如果采用该方式，则难以同时对多机房写入分别做管控，达不到容灾效果。Binlog-->Dsyncer 通常一个 MySQL Table 对应一个转换任务，如果为了写多机房起多个重复的转换任务，则显得有些愚笨。
2. 如果自身业务场景有对同一条记录并发写场景，但写不一定全部来源于 Binlog 的情况下，那全局考虑直写 ES 则更容易遇到写入冲突问题，因为缺乏有序队列的保障。

通过 Flink 搭建 ES 集成系统

1. 通过 MQ 可以更快捷的接入多机房 ES 集群，写入解耦，三机房分别起消费者写入数据， 彼此独立，当出现单机房故障时，只要有可用机房，直接处理读流量切流即可， 容灾方案简单清晰；
2. 网络抖动等问题会导致 ES 暂时性写入失败时，不影响其他集群写入的情况下，RocketMQ 会暂存消息，Flink 会保存消费快照，不断重试直至成功， 更好的保障了数据最终一致性；
3. 多数据源写入能保证全局分区一致性。

1. 依赖了更多组件，会增加全链路数据同步延迟，而 ES 默认的 Refresh 频率是每秒一次，经测试该链路正常情况下数据延迟都是秒级的，不是完全不可接受；
2. 依赖了更多组件，对基础组件的稳定性有更高的要求，RocketMQ 异常，或者 Flink 任务异常都会导致同步链路出现问题，增加一定的业务异常风险。

• Flink 运行环境：首先需要有 Flink 任务的运行环境，通常企业级的 Flink 任务会作为一个 YARN 作业在分布式系统中被调度并分配资源执行，但同时 Flink 也可作为单机进程，亦或搭建一个独立集群运行。
• ES 消息格式：需要约定一种 ES 消息传输格式和序列化方式，一套范式解决所有同步场景，目前流行的序列化方式是 pb 格式或 json 格式，目前我们都是推荐使用 pb 格式的，数据格式 Schema 定义：

syntax = "proto3";

message ESIndexInfo {
    string Name = 1;  // 文档要写入索引的名称或别名
}

enum ESOPType { // 文档写入操作类型
    DELETE = 0; // 删除文档
    INDEX = 1;  // 创建新文档或更新老文档，只能全量更新 (替换老文档)
    UPDATE = 2; // 更新老文档，支持部分更新 (合并老文档)
    UPSERT = 3; // 创建新文档或更新老文档，支持部分更新 (合并老文档)
    CREATE = 4; // 创建新文档，存在时报错丢弃
}

message ESDocAction {
    ESIndexInfo IndexInfo = 1; // 索引信息 (必需)
    ESOPType OPType = 2;       // 操作类型 (必需)
    string ID = 3;             // 文档 ID (可选)
    string Doc = 4;            // 文档内容 (JSON 格式, 删除操作时不需要)
    int64 Version = 5;         // 文档版本 (可选, 大于 0 且操作为 index/create/delete 有效)
    string Routing = 6;        // 文档路由 (可选, 非空有效)
    string Script = 7;         // 文档脚本 (JSON 格式, 操作类型为 update/upsert 有效，但和 Doc 不能同时存在)
}

• Flink 任务必要配置：监听的 RocketMQ Topic 信息，写 ES 集群信息；
• Flink 执行函数：Flink 处理流式消息有流式 SQL 和自定义应用程序两种方式，流式 SQL 约束于本身的一些限制，比如不支持同一个 MQ 有多个索引消息，而自定义编程更加灵活，比如添加各种打点，日志，错误码处理等，推荐该方式；
• Flink 资源配置：JobManager 资源配置，TaskManager 资源配置等等；
• Flink 自定义参数配置：可以自定义一些与应用程序紧密相关的动态配置，方便动态调节 Flink 消费能力，比如：

对比建议