Spark Structured Streaming按环境参数化订阅Kafka Topic问题

代码不生效有3个核心错误：

• 字符串插值语法错误。Scala中只有在字符串前加s前缀时，${变量名}才会被替换为实际变量值，你代码里"test_${envi}"、"/tmp/strlocation_${envi}/AB/checkpointtest1"都是普通字符串，运行时不会替换envi的值，会直接把${envi}当成字符串的一部分，最终订阅的topic、写入的checkpoint路径都是错的。
• getEnvironment变量赋值逻辑错误。你写的getEnvironment = args(1).toString.trim().toUpperCase == "TESTDEVC"是布尔判断，最终getEnvironment只会得到true/false两个布尔值，后面你拿它和"TESTDEVC"这类字符串做相等判断永远不会成立，envi变量会一直是空字符串，根本拿不到传入的环境参数。
• 缺少参数合法性校验。如果作业启动时传入的参数数量不足2个，代码不会抛出明确提示，会直接用空值继续执行，导致后续配置全部错误。

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.functions.col

// 参数解析与校验
if (args.length < 2) {
  throw new RuntimeException("请传入2个启动参数：第1个为环境类型(Q代表QA)，第2个为具体部署环境标识")
}
val isQA: Boolean = args(0).toString.trim().toUpperCase == "Q"
val getEnvironment: String = args(1).toString.trim().toUpperCase

var env = ""
var envi = ""
if (isQA) {
  configManager.setup("test.conf")
  env = "QA"
  envi = getEnvironment match {
    case "TESTDEVC" => "TESTDEVC"
    case "TESTDEVC1" => "TESTDEVC1"
    case "TESTDEVC2" => "TESTDEVC2"
    case _ => throw new RuntimeException(s"未知环境标识: $getEnvironment，仅支持TESTDEVC/TESTDEVC1/TESTDEVC2")
  }
} else {
  throw new RuntimeException("仅支持QA环境启动，请确认第一个启动参数为Q")
}

// 打印解析到的配置，启动时可直接在日志中校验
println(s"=== 作业启动配置 ===")
println(s"当前环境类型: $env")
println(s"订阅Kafka Topic: topicname_${envi}")
println(s"Checkpoint存储路径: /tmp/strlocation_${envi}/AB/checkpointtest1")

val Q_stream = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "Kafka.Server")
  // 字符串前加s前缀，启用插值替换
  .option("subscribe", s"topicname_${envi}")
  .option("startingOffsets", "latest")
  .option("failOnDataLoss", "false")
  .option("kafka.group.id", "myConsumerGroup")
  .load()
  .select(col("key").cast("string").as("key"), col("value").cast("string"))


val value2Stream = Q_stream
  .filter(col("key") === "AB")
  .select(functions.from_json(col("value"), ABSchema).as("value"))
  .select("value.*")

value2Stream.writeStream.format("orc")
.option("metastoreUri", "hive.warehouse.metastoreUri")
// 同样加s前缀
.option("checkpointLocation", s"/tmp/strlocation_${envi}/AB/checkpointtest1")
.option("path", "/tmp/str2/AB")
.trigger(Trigger.ProcessingTime("5 Seconds"))
.partitionBy("jobid")
.start()

• 启动作业时必须按顺序传入2个参数：第一个参数固定传Q走QA环境逻辑，第二个参数传实际环境值，比如spark-submit --class 你的主类名 你的jar包路径 Q TESTDEVC就是订阅topicname_TESTDEVC
• 不同环境的checkpoint路径必须隔离，不要复用同一个目录，否则会出现offset不匹配、流启动失败的问题
• 作业启动时先看日志里打印的=== 作业启动配置 ===部分，确认topic和路径解析正确再观察消费情况

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