从Spark Streaming DataFrame中删除(损坏)不符合模式的行(从Kafka传入的JSON数据)

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder()
  .appName("Spark Streaming Example")
  .master("local[2]")
  .getOrCreate()

// 从Kafka读取JSON数据流
val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic_name")
  .load()

// 将数据流转换为DataFrame
val jsonDF = kafkaDF.selectExpr("CAST(value AS STRING)").select(from_json($"value", schema).as("data"))
val filteredDF = jsonDF.filter(row => isValid(row.getAs[String]("data")))

// 定义一个函数来检查JSON数据是否符合模式
def isValid(json: String): Boolean = {
  // 在这里编写验证逻辑，判断JSON是否符合模式
}

// 输出结果到控制台
val query = filteredDF.writeStream
  .outputMode("append")
  .format("console")
  .start()

query.awaitTermination()