--资源分配，并行度，文件数，task数
spark.dynamicAllocation.enabled true --动态分配
spark.dynamicAllocation.minExecutors 1 --单任务最小分配executor数
spark.dynamicAllocation.maxExecutors 100 --单任务最大分配executor数
spark.executor.cores 1 -- 一个executor的核数
spark.executor.memory 4G -- 一个executor的内存大小
spark.driver.cores 1 -- 一个driver的核数
spark.driver.memory 4G -- 一个driver的内存大小

总的资源：
核数： 100 (executor的核数)+1（driver的核数） = 101 个cpu
内存： 4 * 100 （executor的内存）+ 4 （driver的内存）= 404G

Map阶段：
set spark.sql.files.maxPartitionBytes=128m;
spark.dynamicAllocation.maxExecutors 100 
spark.executor.cores 1
假如只有一个文件，25600M / 128M = 200 个mapTask ==>最多申请100个executor 100个core,至少跑2轮

set spark.sql.files.maxPartitionBytes=128m;
spark.dynamicAllocation.maxExecutors 200 
spark.executor.cores 1
假如只有一个文件，25600M / 128M = 200 个mapTask ==>最多申请200个executor 200个core,至少跑1轮

set spark.sql.files.maxPartitionBytes=256m;
spark.dynamicAllocation.maxExecutors 100 
spark.executor.cores 1
假如只有一个文件，25600M / 128M = 100 个mapTask ==>最多申请100个executor 100个core,至少跑1轮

set spark.sql.files.maxPartitionBytes=128m;
spark.dynamicAllocation.maxExecutors 100 
spark.executor.cores 1
假如只有一个文件，12800M / 128M = 10 个mapTask ==>最多申请10个executor 10个core,至少跑1轮

注意点：
1.无论数据打到多少个task(partition)，任务真正的最大并行度由executor数*core数决定
2.MapTask数无法精确计算，只是估算，大概知道MapTask数 约等于 文件大小 / 128M 就可以
        一个文件只有100M，也做一个task，一个文件切完最后130M，也会是一个MapTask(128 * 1.1=148M的阈值)
        文件有大有小，spark会进行小文件读时合并，将多个小文件读取到一个task
3.资源调度高峰时，并不是你申请100个executor就会给你100个executor
        取决于集群资源是否充足，任务的队列优先级是否高

Reduce节阶段
set spark.sql.shuffle.partitions = 100;
spark.dynamicAllocation.maxExecutors 100
spark.executor.cores 1
100个reduceTask ==> 最多申请100个executor，100个core,至少跑1轮

set spark.sql.shuffle.partitions = 200;
spark.dynamicAllocation.maxExecutors 100
spark.executor.cores 1
200个reduceTask ==> 最多申请100个executor，100个core,至少跑2轮

set spark.sql.shuffle.partitions = 200;
spark.dynamicAllocation.maxExecutors 200
spark.executor.cores 1
200个reduceTask ==> 最多申请200个executor，200个core,至少跑1轮

set spark.sql.shuffle.partitions = 10;
spark.dynamicAllocation.maxExecutors 100
spark.executor.cores 1
10个reduceTask ==> 最多申请10个executor，10个core,至少跑1轮

注意点：
1、reduce数量由参数精准控制（不考虑AQE下）
2、无论数据打到多少个task（partition），任务真正的最大并行度由executor数*core数决定
        并非你 shuffle.partitions，并行度就有多少，set 只是数据打散的分区数
        既然不能提高并行度，那么设置大点有什么好处？
                （1）、数据量小时没有好处 ，每个task跑的都很快，但因为executor数限制，要跑多轮，每个task资源的申请，分配，回收都需要时间，尤其调度高峰
                （2）、数据量大时有好处，合理利用资源，减少shuffle的溢写，让任务跑的更快
3、shuffle分发规则：
hash分发：key.hash() % reduce数，得到对应的reduceid ==> group by a,b join ,distribute by
roundRobin分发：轮询分发   ==> /*+ repartition(5) */


-- 假设
set spark.sql.files.maxPartitionBytes = 128m;
set spark.sql.shuffle.partitions = 200;
spark.dynamicAllocation.maxExecutors 100
spark.executor.cores 1
只有一个可文件，25600M / 128 = 200个mapTask

--静态分区
--只有map阶段，200个MapTask，无Reduce
--文件数=mapTask数量=200个文件
insert overwrite tar_talbe partitions by(day_id='2024-07-01') select * from src table

--repartition,200个MapTask,2个Reduce
--文件数=repartition的数量=2个文件
insert overwrite tar_talbe partitions by(day_id='2024-07-01') select /*+ repartition(2) */ from src table

--distribute by,200个MapTask,200个Reduce,hash分发:key.hash() % reduce数
--文件数 = key.hash() % reduce数(200)的枚举数量
insert overwrite tar_talbe partitions by(day_id='2024-07-01') 
select * from src table
distribute by rand() --200个
distribute by 1 --1个
distribute by ceil(rand()) * 5 --1~5个,取决于 key.hash() % reduce数（200）的枚举数量


--动态分区
--repartition ，200个MapTask,2个Reduce
--文件数=2(repartition数) * day_id分区数
insert overwrite tar_talbe partitions by (day_id)
select /*+ repartition(2) */
*,
day_id  --day_id 为分区键
from str_table 

--动态分区
--distribute by, 200个MapTask, 200个Reduce, hash分发: key.hash()%reduce数
insert overwrite tar_talbe partitions by(day_id) 
select 
        *
        ,day_id  --day_id 为分区键 --3个分区值
from src table
distribute by rand() --文件数= 200 * day_id分区数（3） = 600
distribute by day_id --文件数= day_id数 
distribute by (day_id,ceil(rand()) * 5) --文件数= day_id数 * (1~5)