Spark如何设置全局变量 实现多环境下Kafka主题动态适配

核心实现逻辑：抽离独立公共配置文件统一管理Kafka参数，作业提交时传入当前环境标识，配置文件自动匹配对应Kafka主题，所有业务代码直接引用公共配置项即可，无需重复传参或硬编码。

1. 编写统一公共配置文件kafkaCom #

单独创建kafkaCom文件存放所有Kafka相关公共参数，所有业务代码可直接引用该文件内的配置，实现全局变量效果。
配置文件内自动读取提交时传入的环境标识，通过预定义的映射关系匹配对应主题，新增环境仅需修改该文件的映射表即可，无需调整业务代码。

Scala/Java 版本示例 #

// kafkaCom.scala 全局Kafka公共配置
import org.apache.spark.sql.SparkSession

object kafkaCom {
  // 读取当前作业运行环境，未配置环境直接抛出异常避免误连
  private val currentEnv: String = SparkSession.getActiveSession match {
    case Some(spark) => spark.conf.get("spark.run.env", "").toLowerCase
    case None => throw new IllegalStateException("SparkSession未初始化，无法加载Kafka配置")
  }

  // 环境与Kafka主题后缀映射表，新增环境直接追加对应键值对
  private val topicSuffixMapping = Map(
    "test1" -> "one",
    "test2" -> "two",
    "test3" -> "three"
  )

  // 全局暴露的公共配置项，业务代码直接引用即可
  val kafkaTargetTopic: String = s"test_${topicSuffixMapping.getOrElse(currentEnv, throw new IllegalArgumentException(s"环境[$currentEnv]未配置对应Kafka主题"))}"
  val kafkaBootstrapServers: String = s"kafka-cluster-$currentEnv:9092"
  val commonKafkaParams: Map[String, String] = Map(
    "failOnDataLoss" -> "false",
    "maxOffsetsPerTrigger" -> "100000",
    "startingOffsets" -> "latest"
  )
}

PySpark 版本示例 #

# kafkaCom.py 全局Kafka公共配置
from pyspark.sql import SparkSession

def _get_init_spark():
    try:
        return SparkSession.builder.getOrCreate()
    except Exception as e:
        raise RuntimeError("SparkSession未初始化，无法加载Kafka配置") from e

# 环境与Kafka主题后缀映射表，新增环境直接追加对应键值对
_TOPIC_SUFFIX_MAPPING = {
    "test1": "one",
    "test2": "two",
    "test3": "three"
}

_spark = _get_init_spark()
_current_env = _spark.conf.get("spark.run.env", "").lower()
if _current_env not in _TOPIC_SUFFIX_MAPPING:
    raise ValueError(f"环境[{_current_env}]未配置对应Kafka主题")

# 全局暴露的公共配置项，业务代码直接import即可使用
KAFKA_TARGET_TOPIC = f"test_{_TOPIC_SUFFIX_MAPPING[_current_env]}"
KAFKA_BOOTSTRAP_SERVERS = f"kafka-cluster-{_current_env}:9092"
COMMON_KAFKA_PARAMS = {
    "failOnDataLoss": "false",
    "maxOffsetsPerTrigger": "100000",
    "startingOffsets": "latest"
}

2. 作业提交时传入环境参数 #

不同环境提交作业时，通过spark-submit的--conf参数传入当前环境标识即可，无需修改任何代码：

# TEST1环境提交示例，自动匹配test_one主题
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --conf spark.run.env=test1 \
  # 其余提交参数（jar包/py文件、资源配置等）
  your_application_code.jar

TEST2环境提交时仅需将spark.run.env参数值改为test2，配置会自动匹配test_two主题，TEST3同理。如果配合CI/CD流水线，可直接在各环境的提交脚本中预置该参数，完全无需人工干预。

3. 业务代码引用全局配置 #

所有需要使用Kafka参数的业务代码文件，直接导入kafkaCom中的公共配置项即可，无需重复定义或传参：

Scala 业务代码示例 #

// 任意业务代码文件直接导入公共配置
import kafkaCom._

val kafkaStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaBootstrapServers)
  .option("subscribe", kafkaTargetTopic)
  .options(commonKafkaParams)
  .load()

PySpark 业务代码示例 #

# 任意业务代码文件直接导入公共配置
from kafkaCom import KAFKA_TARGET_TOPIC, KAFKA_BOOTSTRAP_SERVERS, COMMON_KAFKA_PARAMS

kafka_stream_df = spark.readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS) \
  .option("subscribe", KAFKA_TARGET_TOPIC) \
  .options(**COMMON_KAFKA_PARAMS) \
  .load()

注意事项 #

• 不要使用广播变量实现这类配置：广播变量用于Executor端分发大体积只读变量，Driver端公共配置直接通过对象/模块引用的方式实现最稳妥，不会触发序列化问题。
• 禁止在代码中硬编码环境标识：所有环境参数通过提交时传入，避免人为修改代码导致的环境错连问题。
• 所有Kafka相关的公共配置统一收敛在kafkaCom文件维护，不要在业务代码中散落硬编码的Kafka参数，后续变更仅需修改这一个文件即可。

内容的提问来源于stack exchange，提问作者dataeng