当前位置：首页 > news >正文

Flink 生产问题（数据倾斜）

news 来源：原创 2024/7/5 1:01:18

Flink 生产问题（数据倾斜）

问题概述

任务节点频繁出现反压，但是增加并行度后并不能解决问题；
部分节点出现 OOM 异常，原因是大量的数据集中在某个节点上，导致该节点内存被爆，任务失败重启。

产生数据倾斜的原因：

业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；
技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。

解决问题思路：

业务上要尽量避免热点 key 的设计，例如可以把北京、上海等热点城市分成不同的区域，并进行单独处理；
技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；。

数据倾斜场景案例和解决方案

数据倾斜场景：

统计各省下单次数（北京、上海等几个城市的订单量远远超过其他地区）

解决思路（二次聚合）：

首先把分组的 key 打散，加随机数前缀；
对打散后的数据进行聚合；
把打散的 key 去除随机树前缀还原为真正的 key；
二次 KeyBy 进行结果统计，然后输出。

Java 代码如下：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.Random;public class OrderCountByProvince {public static void main(String[] args) throws Exception {// 创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 模拟输入数据流：订单 (省份, 下单次数)DataStream<Tuple2<String, Integer>> orders = env.fromElements(Tuple2.of("Beijing", 1),Tuple2.of("Shanghai", 1),Tuple2.of("Guangdong", 1),Tuple2.of("Beijing", 1),Tuple2.of("Shanghai", 1),Tuple2.of("Guangdong", 1),Tuple2.of("Beijing", 1),Tuple2.of("Shanghai", 1));// 添加随机前缀，缓解省份热点问题DataStream<Tuple2<String, Integer>> ordersWithPrefix = orders.map(new AddRandomPrefix());// 按带有前缀的省份进行分区统计DataStream<Tuple2<String, Integer>> orderCountsWithPrefix = ordersWithPrefix.keyBy(order -> order.f0).reduce(new ReduceFunction<Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) {return Tuple2.of(value1.f0, value1.f1 + value2.f1);}});// 移除随机前缀，合并前缀后的结果DataStream<Tuple2<String, Integer>> orderCounts = orderCountsWithPrefix.map(new RemoveRandomPrefix()).keyBy(order -> order.f0).reduce(new ReduceFunction<Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) {return Tuple2.of(value1.f0, value1.f1 + value2.f1);}});// 打印结果orderCounts.print();// 执行任务env.execute("Order Count By Province with Hotspot Mitigation");}// 添加随机前缀的 MapFunctionpublic static class AddRandomPrefix implements MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {private static final Random random = new Random();@Overridepublic Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {String randomPrefix = String.valueOf(random.nextInt(10));return Tuple2.of(randomPrefix + "-" + value.f0, value.f1);}}// 移除随机前缀的 MapFunctionpublic static class RemoveRandomPrefix implements MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {@Overridepublic Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {String province = value.f0.split("-", 2)[1];return Tuple2.of(province, value.f1);}}
}