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

news/2025/4/1 1:18:56/

流式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的实时数据同步，同时处理表之间的关联关系。


http://www.ppmy.cn/news/1583242.html

相关文章

第 4 章 | Solidity安全 权限控制漏洞全解析

第 4 章 | Solidity安全 权限控制漏洞全解析

来了&#xff01;我们继续深入安全审计核心模块——本章聚焦另一类高发且致命的漏洞&#xff1a; &#x1f510; 第 4 章 | 权限控制漏洞全解析 ——从 tx.origin 到未初始化 owner&#xff0c;一步走错&#xff0c;合约归人 ✅ 本章导读 智能合约没有“登录系统”&#xff0c…
阅读更多...
人工智能 - SpatialLM：三维空间理解的开源大语言模型

人工智能 - SpatialLM：三维空间理解的开源大语言模型

项目概述 SpatialLM 是由杭州群核科技（“杭州六小龙”之一）于 GTC 2025 大会上开源的三维空间理解大语言模型。该项目旨在通过多模态数据处理能力，将非结构化的三维几何数据（如点云、视频）转化为机器可读的结构化场景描述，为机器人导航、建筑设计、AR/VR 等领域提供高层…
阅读更多...
树莓派5-GPIO和40针引脚

树莓派5-GPIO和40针引脚

1.树莓派5引脚图 2.GPIO 引脚作用 (1) 电压 板上有两个 5V 引脚和两个 3.3V 引脚&#xff0c;以及一些不可配置的接地引脚 (0V)。其余引脚均为通用 3.3V 引 脚&#xff0c;这意味着输出设置为 3.3V&#xff0c;输入可接 3.3V。 (2) 输出 指定为输出引脚的 GPIO 引脚可设置为…
阅读更多...
### Java二维字符矩阵输入解析：正确读取由0和1组成的矩阵

### Java二维字符矩阵输入解析：正确读取由0和1组成的矩阵

在解决LeetCode等编程平台上的算法问题时&#xff0c;正确处理输入数据是解题的第一步。本文以Java语言为例&#xff0c;详细讲解如何正确读取由0和1组成的二维字符矩阵&#xff0c;并解决输入过程中可能遇到的换行符问题。 --- #### **问题背景** 题目要求从输入中读取一个二…
阅读更多...
开源大模型使用总结

开源大模型使用总结

文章目录 OpenRouterGroqCursor直接支持设置Geminivscode插件&#xff1a;Roo code硅基流动嵌入式模型 火山引擎Gemini多模态AI应用Hugging face参考 OpenRouter 地址 1.创建API key 2.在cursor中配置API 在cursor中增加模型 Groq 地址 1.支持的模型&#xff1a; 2.…
阅读更多...
基于单片机的农作物自动灌溉系统

基于单片机的农作物自动灌溉系统

标题:基于单片机的农作物自动灌溉系统 内容:1.摘要 本研究针对传统农作物灌溉方式效率低、水资源浪费严重等问题&#xff0c;旨在设计一种基于单片机的农作物自动灌溉系统。采用土壤湿度传感器实时采集土壤湿度数据&#xff0c;将数据传输至单片机进行处理&#xff0c;单片机根…
阅读更多...
各类神经网络学习：（四）RNN 循环神经网络（下集），pytorch 版的 RNN 代码编写

各类神经网络学习：（四）RNN 循环神经网络（下集），pytorch 版的 RNN 代码编写

上一篇下一篇RNN&#xff08;中集&#xff09;待编写 代码详解 pytorch 官网主要有两个可调用的模块&#xff0c;分别是 nn.RNNCell 和 nn.RNN &#xff0c;下面会进行详细讲解。 RNN 的同步多对多、多对一、一对多等等结构都是由这两个模块实现的&#xff0c;只需要将对输入…
阅读更多...
单片机 - RAM 与内存、ROM 与硬盘 之间的详细对比总结

单片机 - RAM 与内存、ROM 与硬盘 之间的详细对比总结

RAM 与 内存 RAM&#xff08;Random Access Memory&#xff0c;随机存取存储器&#xff09; 和 内存 这两个术语通常是 同义词&#xff0c;即 内存 常常指的就是 RAM。 1. RAM&#xff08;内存&#xff09; 定义&#xff1a;RAM 是计算机中的 主存储器&#xff0c;用于临时存…
阅读更多...
最新文章

Copyright @ 2022~2023