Databricks Spark2.4.5单JVM仅允许一个StreamingContext报错如何解决？

核心实现逻辑 #

JVM单实例限制下，先主动清理已存在的活跃StreamingContext，且停止时不终止当前Spark Context，再创建新实例启动即可，无需重启集群或Spark Context。

可直接复用的代码模板 #

Scala 版本 #

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds

// 第一步：清理已存在的活跃流上下文
val activeSsc = StreamingContext.getActive()
if (activeSsc != null) {
  // 第一个false表示不停止当前SparkContext，第二个true表示优雅等待存量微批次处理完成
  activeSsc.stop(stopSparkContext = false, stopGracefully = true)
  // 等待旧上下文完全释放资源，可根据自身微批次时长调整等待时间
  Thread.sleep(3000)
}

// 第二步：创建新的流上下文，直接复用当前已有的SparkContext（即sc变量）
val ssc = new StreamingContext(sc, Seconds(10))

// 此处写入你的流处理业务逻辑
// val lines = ssc.socketTextStream(...)
// ...

// 启动新流作业
ssc.start()
ssc.awaitTermination()

Java 版本 #

import org.apache.spark.streaming.StreamingContext;
import org.apache.spark.streaming.Durations;

// 第一步：清理已存在的活跃流上下文
StreamingContext activeSsc = StreamingContext.getActive();
if (activeSsc != null) {
    activeSsc.stop(false, true);
    Thread.sleep(3000);
}

// 第二步：创建新的流上下文，复用当前JavaSparkContext（即jsc变量）
StreamingContext ssc = new StreamingContext(jsc, Durations.seconds(10));

// 此处写入你的流处理业务逻辑
// JavaReceiverInputDStream<String> lines = ssc.socketTextStream(...)
// ...

// 启动新流作业
ssc.start();
ssc.awaitTermination();

注意事项 #

• 测试阶段不需要优雅停止的话，可将stopGracefully设为false，加快旧上下文释放速度
• 未开启checkpoint的情况下，新启动的流作业不会继承旧作业的消费偏移量，生产环境建议补充checkpoint配置避免数据丢失
• 若流作业用到了RDD持久化、临时表，可在停止旧上下文后手动调用sc.clearPersistentRDDs()等方法清理冗余资源

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