无法用Spark结构化流覆盖 "spark.sql.shuffle.partitions "的默认值

2 人关注

我想在代码中直接覆盖 spark.sql.shuffle.partitions 参数。

val sparkSession = SparkSession
  .builder()
  .appName("SPARK")
  .getOrCreate()
sparkSession.conf.set("spark.sql.shuffle.partitions", 2)
但这个设置并不生效，因为在日志中我得到了以下警告信息。
WARN  OffsetSeqMetadata:66 - Updating the value of conf 'spark.sql.shuffle.partitions' in current session from '2' to '200'.
而在spark-submit shell中传递的相同参数却能工作。
#!/bin/bash
/app/spark-2/bin/spark-submit \
--queue root.dev \
--master yarn \
--deploy-mode cluster \
--driver-memory 5G \
--executor-memory 4G \
--executor-cores 2 \
--num-executors 4 \
--conf spark.app.name=SPARK \
--conf spark.executor.memoryOverhead=2048 \
--conf spark.yarn.maxAppAttempts=1 \
--conf spark.sql.shuffle.partitions=2 \
--class com.dev.MainClass
有什么想法吗？


         
          
           scala


         
          
           apache-spark


         
          
           spark-structured-streaming


          
           
            
             
             
              Michael Heil
             
            
            
             发布于
             
             2021-05-08


          
           
            
             在你的Spark结构化流作业的检查点文件中，一些
             
              sparkSession
             
             的配置被存储。
            
            
             例如，在 "offset "文件夹中，最新批次的内容可能看起来像。
            
            {
            
             "batchWatermarkMs"
            
            :
            
             0
            
            ,
            
             "batchTimestampMs"
            
            :
            
             1619782960476
            
            ,
            
             "conf"
            
            :{
            
             "spark.sql.streaming.stateStore.providerClass"
            
            :
            
             "org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider"
            
            ,
            
             "spark.sql.streaming.join.stateFormatVersion"
            
            :
            
             "2"
            
            ,
            
             "spark.sql.streaming.stateStore.compression.codec"
            
            :
            
             "lz4"
            
            ,
            
             "spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion"
            
            :
            
             "2"
            
            ,
            
             "spark.sql.streaming.multipleWatermarkPolicy"
            
            :
            
             "min"
            
            ,
            
             "spark.sql.streaming.aggregation.stateFormatVersion"
            
            :
            
             "2"
            
            ,
            
             "spark.sql.shuffle.partitions"
            
            :
            
             "200"
            
            }}