许多流媒体源的检查点

Question

我正在使用 zeppelin，我在 spark streaming 中从许多来源读取了许多文件，如下所示：

    val var1 = spark
  .readStream
  .schema(var1_raw)      
  .option("sep", ",")  
  .option("mode", "PERMISSIVE")  
  .option("maxFilesPerTrigger", 100)
  .option("treatEmptyValuesAsNulls", "true")
  .option("newFilesOnly", "true") 
  .csv(path_var1 )    


val chekpoint_var1 =  var1
    .writeStream
    .format("csv") 
    .option("checkpointLocation", path_checkpoint_var1) 
    .option("Path",path_checkpoint )  
    .option("header", true)  
    .outputMode("Append")
    .queryName("var1_backup")
    .start().awaitTermination()


val var2 = spark
    .readStream
  .schema(var2_raw)      
  .option("sep", ",")  
  .option("mode", "PERMISSIVE")  //
  .option("maxFilesPerTrigger", 100)
  .option("treatEmptyValuesAsNulls", "true") 
  .option("newFilesOnly", "true") 
  .csv(path_var2 )   

val chekpoint_var2 =  var2
    .writeStream
    .format("csv") 
    .option("checkpointLocation", path_checkpoint_var2)   //
    .option("path",path_checkpoint_2 )  
    .option("header", true)  
    .outputMode("Append")
    .queryName("var2_backup")
    .start().awaitTermination()

当我运行这份工作时，我收到了这条消息： java.lang.IllegalArgumentException：无法启动名称为 var1_backup 的查询，因为名称为

的查询已处于活动状态

*****************解决方案******************

val spark = SparkSession
    .builder
    .appName("test")
    .config("spark.local", "local[*]")
    .getOrCreate()
spark.sparkContext.setCheckpointDir(path_checkpoint)

在我调用数据帧上的检查点函数后

Answer 1

*****************解决方案******************

val spark = SparkSession
    .builder
    .appName("test")
    .config("spark.local", "local[*]")
    .getOrCreate()
spark.sparkContext.setCheckpointDir(path_checkpoint)

在我调用数据帧上的检查点函数后

许多流媒体源的检查点

Checkpoint for many streaming source

scala

apache-spark

apache-spark-sql

apache-zeppelin

spark-structured-streaming