1、测试反压：

        1、反压：

指的是下游消费数据的速度比上游产生数据的速度要小时会出现反压，下游导致上游的Task反压。

        2、测试反压：使用的是DataGen

CREATE TABLE words (word STRING
) WITH ('connector' = 'datagen','rows-per-second'='100000', -- 每秒随机生成的数据量'fields.word.length'='4'
);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-second'='1000000','fields.word.length'='5') */
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.enabled'='true';-- 批次的时间
set 'table.exec.mini-batch.allow-latency'='5 s';-- 批次大小
set 'table.exec.mini-batch.size'='5000';-- 开启预聚合
set 'table.optimizer.agg-phase-strategy'='TWO_PHASE';