流式ETL配置指南：从MySQL到Elasticsearch的实时数据同步

流式ETL配置指南：从MySQL到Elasticsearch的实时数据同步

场景介绍

假设您运营一个电商平台，需要将MySQL数据库中的订单、用户和产品信息实时同步到Elasticsearch，以支持实时搜索、分析和仪表盘展示。传统的批处理ETL无法满足实时性要求，因此我们将使用Flink CDC构建流式ETL管道。

前提条件

1. MySQL数据库 (作为数据源)
2. Elasticsearch (作为目标系统)
3. Flink环境 (处理引擎)
4. Java开发环境

步骤一：环境准备

1.1 准备MySQL环境

-- 创建数据库
CREATE DATABASE IF NOT EXISTS shop;
USE shop;-- 创建用户表
CREATE TABLE users (id INT PRIMARY KEY,name VARCHAR(100),email VARCHAR(100),create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);-- 创建产品表
CREATE TABLE products (id INT PRIMARY KEY,name VARCHAR(200),price DECIMAL(10,2),stock INT,category VARCHAR(100)
);-- 创建订单表
CREATE TABLE orders (id INT PRIMARY KEY,user_id INT,order_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,status VARCHAR(20),total_amount DECIMAL(10,2),FOREIGN KEY (user_id) REFERENCES users(id)
);-- 创建订单详情表
CREATE TABLE order_items (id INT PRIMARY KEY,order_id INT,product_id INT,quantity INT,price DECIMAL(10,2),FOREIGN KEY (order_id) REFERENCES orders(id),FOREIGN KEY (product_id) REFERENCES products(id)
);-- 插入一些测试数据
INSERT INTO users VALUES (1, '张三', 'zhangsan@example.com', '2023-01-01 10:00:00');
INSERT INTO products VALUES (101, 'iPhone 14', 5999.00, 100, '电子产品');
INSERT INTO orders VALUES (1001, 1, '2023-01-05 14:30:00', '已完成', 5999.00);
INSERT INTO order_items VALUES (10001, 1001, 101, 1, 5999.00);

确保MySQL已开启binlog，编辑MySQL配置文件：

[mysqld]
server-id=1
log-bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL

1.2 准备Elasticsearch环境

创建索引映射：

PUT /shop_orders
{"mappings": {"properties": {"order_id": { "type": "keyword" },"user_id": { "type": "keyword" },"user_name": { "type": "keyword" },"user_email": { "type": "keyword" },"order_time": { "type": "date" },"status": { "type": "keyword" },"total_amount": { "type": "double" },"items": {"type": "nested","properties": {"product_id": { "type": "keyword" },"product_name": { "type": "text" },"quantity": { "type": "integer" },"price": { "type": "double" },"category": { "type": "keyword" }}}}}
}

步骤二：创建Flink流式ETL项目

2.1 创建Maven项目

pom.xml文件配置：

<dependencies><!-- Flink核心依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>1.17.0</version></dependency><!-- Flink CDC连接器 --><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.3.0</version></dependency><!-- Elasticsearch连接器 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch7</artifactId><version>1.17.0</version></dependency><!-- JSON处理 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>1.17.0</version></dependency><dependency><groupId>com.google.code.gson</groupId><artifactId>gson</artifactId><version>2.9.0</version></dependency>
</dependencies>

2.2 实现ETL主程序

创建MySQLToElasticsearchETL.java文件：

import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;public class MySQLToElasticsearchETL {public static void main(String[] args) throws Exception {// 1. 设置Flink执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);  // 开发环境设置为1，生产环境根据需要调整env.enableCheckpointing(60000);  // 每60秒做一次检查点// 2. 配置MySQL CDC源MySqlSource<String> userSource = MySqlSource.<String>builder().hostname("localhost").port(3306).databaseList("shop").tableList("shop.users").username("root").password("yourpassword").deserializer(new JsonDebeziumDeserializationSchema()).startupOptions(StartupOptions.initial()).build();MySqlSource<String> productSource = MySqlSource.<String>builder().hostname("localhost").port(3306).databaseList("shop").tableList("shop.products").username("root").password("yourpassword").deserializer(new JsonDebeziumDeserializationSchema()).startupOptions(StartupOptions.initial()).build();MySqlSource<String> orderSource = MySqlSource.<String>builder().hostname("localhost").port(3306).databaseList("shop").tableList("shop.orders").username("root").password("yourpassword").deserializer(new JsonDebeziumDeserializationSchema()).startupOptions(StartupOptions.initial()).build();MySqlSource<String> orderItemSource = MySqlSource.<String>builder().hostname("localhost").port(3306).databaseList("shop").tableList("shop.order_items").username("root").password("yourpassword").deserializer(new JsonDebeziumDeserializationSchema()).startupOptions(StartupOptions.initial()).build();// 3. 创建数据流DataStream<String> userStream = env.fromSource(userSource,WatermarkStrategy.noWatermarks(),"User CDC Source");DataStream<String> productStream = env.fromSource(productSource,WatermarkStrategy.noWatermarks(),"Product CDC Source");DataStream<String> orderStream = env.fromSource(orderSource,WatermarkStrategy.noWatermarks(),"Order CDC Source");DataStream<String> orderItemStream = env.fromSource(orderItemSource,WatermarkStrategy.noWatermarks(),"OrderItem CDC Source");// 4. 数据转换与关联// 用户缓存Map<Integer, Map<String, Object>> userCache = new HashMap<>();userStream.map(json -> {JsonObject jsonObject = JsonParser.parseString(json).getAsJsonObject();JsonObject after = jsonObject.getAsJsonObject("after");if (after != null) {int userId = after.get("id").getAsInt();Map<String, Object> userInfo = new HashMap<>();userInfo.put("name", after.get("name").getAsString());userInfo.put("email", after.get("email").getAsString());userCache.put(userId, userInfo);}return json;});// 产品缓存Map<Integer, Map<String, Object>> productCache = new HashMap<>();productStream.map(json -> {JsonObject jsonObject = JsonParser.parseString(json).getAsJsonObject();JsonObject after = jsonObject.getAsJsonObject("after");if (after != null) {int productId = after.get("id").getAsInt();Map<String, Object> productInfo = new HashMap<>();productInfo.put("name", after.get("name").getAsString());productInfo.put("price", after.get("price").getAsDouble());productInfo.put("category", after.get("category").getAsString());productCache.put(productId, productInfo);}return json;});// 订单与订单项关联Map<Integer, List<Map<String, Object>>> orderItemsCache = new HashMap<>();orderItemStream.map(json -> {JsonObject jsonObject = JsonParser.parseString(json).getAsJsonObject();JsonObject after = jsonObject.getAsJsonObject("after");if (after != null) {int orderId = after.get("order_id").getAsInt();int productId = after.get("product_id").getAsInt();Map<String, Object> itemInfo = new HashMap<>();itemInfo.put("product_id", productId);itemInfo.put("quantity", after.get("quantity").getAsInt());itemInfo.put("price", after.get("price").getAsDouble());// 添加产品信息if (productCache.containsKey(productId)) {itemInfo.put("product_name", productCache.get(productId).get("name"));itemInfo.put("category", productCache.get(productId).get("category"));}if (!orderItemsCache.containsKey(orderId)) {orderItemsCache.put(orderId, new ArrayList<>());}orderItemsCache.get(orderId).add(itemInfo);}return json;});// 处理订单并关联用户和订单项SingleOutputStreamOperator<Map<String, Object>> enrichedOrderStream = orderStream.map(new MapFunction<String, Map<String, Object>>() {@Overridepublic Map<String, Object> map(String json) throws Exception {JsonObject jsonObject = JsonParser.parseString(json).getAsJsonObject();JsonObject after = jsonObject.getAsJsonObject("after");String op = jsonObject.get("op").getAsString();Map<String, Object> orderInfo = new HashMap<>();// 只处理插入和更新事件if ("c".equals(op) || "u".equals(op)) {int orderId = after.get("id").getAsInt();int userId = after.get("user_id").getAsInt();orderInfo.put("order_id", orderId);orderInfo.put("user_id", userId);orderInfo.put("order_time", after.get("order_time").getAsString());orderInfo.put("status", after.get("status").getAsString());orderInfo.put("total_amount", after.get("total_amount").getAsDouble());// 关联用户信息if (userCache.containsKey(userId)) {orderInfo.put("user_name", userCache.get(userId).get("name"));orderInfo.put("user_email", userCache.get(userId).get("email"));}// 关联订单项if (orderItemsCache.containsKey(orderId)) {orderInfo.put("items", orderItemsCache.get(orderId));}}return orderInfo;}});// 5. 配置Elasticsearch接收器List<HttpHost> httpHosts = new ArrayList<>();httpHosts.add(new HttpHost("localhost", 9200, "http"));ElasticsearchSink.Builder<Map<String, Object>> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts,(request, context, element) -> {if (element.containsKey("order_id")) {request.index("shop_orders").id(element.get("order_id").toString()).source(element);}});// 配置批量写入esSinkBuilder.setBulkFlushMaxActions(1);  // 每条记录立即写入，生产环境可以调大esSinkBuilder.setBulkFlushInterval(1000);  // 每秒刷新一次// 6. 写入ElasticsearchenrichedOrderStream.addSink(esSinkBuilder.build());// 7. 执行作业env.execute("MySQL to Elasticsearch ETL Job");}
}

步骤三：部署和运行

3.1 编译打包

使用Maven打包：

mvn clean package

3.2 提交到Flink集群

flink run -c MySQLToElasticsearchETL target/your-jar-file.jar

3.3 验证数据同步

在Elasticsearch中查询数据：

curl -X GET "localhost:9200/shop_orders/_search?pretty"

关键点和注意事项

1. 数据一致性：

   • 确保开启Flink的检查点机制，实现exactly-once语义
   • 合理设置检查点间隔，平衡一致性和性能

2. 状态管理：

   • 在上述例子中，我们在内存中维护了用户和产品的缓存，生产环境应使用Flink的状态API
   • 考虑状态大小和清理策略，避免状态无限增长

3. 表关联策略：

   • 上述示例使用了简化的表关联方式
   • 生产环境可以考虑使用Flink SQL或异步I/O进行优化

4. 性能优化：

   • 调整并行度以匹配业务需求
   • 设置合适的批处理大小和间隔
   • 监控反压（backpressure）情况

5. 错误处理：

   • 添加错误处理逻辑，处理数据格式异常
   • 实现重试机制，应对临时网络故障
   • 考虑死信队列（DLQ）来处理无法处理的消息

6. 监控和告警：

   • 接入Prometheus和Grafana监控Flink作业
   • 设置关键指标告警，如延迟、失败次数等

7. 扩展性考虑：

   • 设计时考虑表结构变更的处理方式
   • 为未来增加新数据源或新目标系统预留扩展点

扩展功能

基于这个基础架构，您可以进一步实现：

1. 增量更新优化：只同步变更字段，减少网络传输
2. 历史数据回溯：支持从特定时间点重新同步数据
3. 数据转换：增加复杂的业务计算逻辑
4. 数据过滤：根据业务规则过滤不需要的数据
5. 多目标写入：同时将数据写入Elasticsearch和其他系统如Kafka

这个完整的方案展示了如何使用Flink CDC构建一个端到端的流式ETL系统，实现从MySQL到Elasticsearch的实时数据同步，同时处理表之间的关联关系。