流式ETL配置指南：从MySQL到Elasticsearch的实时数据同步
场景介绍
假设您运营一个电商平台，需要将MySQL数据库中的订单、用户和产品信息实时同步到Elasticsearch，以支持实时搜索、分析和仪表盘展示。传统的批处理ETL无法满足实时性要求，因此我们将使用Flink CDC构建流式ETL管道。
前提条件
- MySQL数据库 (作为数据源)
- Elasticsearch (作为目标系统)
- Flink环境 (处理引擎)
- 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>>() {
@Override
public 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. 写入Elasticsearch
enrichedOrderStream.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"
关键点和注意事项
数据一致性：
- 确保开启Flink的检查点机制，实现exactly-once语义
- 合理设置检查点间隔，平衡一致性和性能
状态管理：
- 在上述例子中，我们在内存中维护了用户和产品的缓存，生产环境应使用Flink的状态API
- 考虑状态大小和清理策略，避免状态无限增长
表关联策略：
- 上述示例使用了简化的表关联方式
- 生产环境可以考虑使用Flink SQL或异步I/O进行优化
性能优化：
- 调整并行度以匹配业务需求
- 设置合适的批处理大小和间隔
- 监控反压（backpressure）情况
错误处理：
- 添加错误处理逻辑，处理数据格式异常
- 实现重试机制，应对临时网络故障
- 考虑死信队列（DLQ）来处理无法处理的消息
监控和告警：
- 接入Prometheus和Grafana监控Flink作业
- 设置关键指标告警，如延迟、失败次数等
扩展性考虑：
- 设计时考虑表结构变更的处理方式
- 为未来增加新数据源或新目标系统预留扩展点
扩展功能
基于这个基础架构，您可以进一步实现：
- 增量更新优化：只同步变更字段，减少网络传输
- 历史数据回溯：支持从特定时间点重新同步数据
- 数据转换：增加复杂的业务计算逻辑
- 数据过滤：根据业务规则过滤不需要的数据
- 多目标写入：同时将数据写入Elasticsearch和其他系统如Kafka
这个完整的方案展示了如何使用Flink CDC构建一个端到端的流式ETL系统，实现从MySQL到Elasticsearch的实时数据同步，同时处理表之间的关联关系。