制作网站的模板,免费注册网站怎么做链接,直播软件app平台大全,北京网站制作网络推广公司1、测试反压#xff1a; 1、反压#xff1a;
指的是下游消费数据的速度比上游产生数据的速度要小时会出现反压#xff0c;下游导致上游的Task反压。 2、测试反压#xff1a;使用的是DataGen
CREATE TABLE words (word STRING
) WITH (connector datagen,rows-per-second…1、测试反压 1、反压
指的是下游消费数据的速度比上游产生数据的速度要小时会出现反压下游导致上游的Task反压。 2、测试反压使用的是DataGen
CREATE TABLE words (word STRING
) WITH (connector datagen,rows-per-second100000, -- 每秒随机生成的数据量fields.word.length4
);CREATE TABLE blackhole_table (wprd STRING,num BIGINT
) WITH (connector blackhole
);insert into blackhole_table
select
word,
count(1) as num from
words /* OPTIONS(rows-per-second1000000,fields.word.length5) */
group by word;
2、解决反压 1、增加资源
--1、增加并行度一个并行度对应一个slot
SET parallelism.default 2;--2、增加内存
-- 如果状态太大内存放不下导致的反压可以通过增加内存解决
-- -tm : taskmanager的内存
-- -jm : jobmanager的内存
yarn-session.sh -tm 4G -d 2、微批处理和预聚合
1、MiniBatch 聚合的核心思想是将一组输入的数据缓存在聚合算子内部的缓冲区中。当输入的数据被触发处理时每个 key 只需一个操作即可访问状态。这样可以大大减少状态开销并获得更好的吞吐量。但是这可能会增加一些延迟因为它会缓冲一些记录而不是立即处理它们。这是吞吐量和延迟之间的权衡。
2、预聚合
数据流中的记录可能会倾斜因此某些聚合算子的实例必须比其他实例处理更多的记录这会产生热点问题。本地聚合可以将一定数量具有相同 key 的输入数据累加到单个累加器中。全局聚合将仅接收 reduce 后的累加器而不是大量的原始输入数据。这可以大大减少网络 shuffle 和状态访问的成本。每次本地聚合累积的输入数据量基于 mini-batch 间隔。这意味着 local-global 聚合依赖于启用了 mini-batch 优化。 3、开启微批处理和预聚合可以减少shuffle过程中传输的数据量减轻下游算子计算的压力
-- 开启微批处理
set table.exec.mini-batch.enabledtrue;-- 批次的时间
set table.exec.mini-batch.allow-latency5 s;-- 批次大小
set table.exec.mini-batch.size5000;-- 开启预聚合
set table.optimizer.agg-phase-strategyTWO_PHASE;
