在spark结构化流媒体中从kafka/json数据源写入损坏的数据

Question

在火花批处理作业中，我通常将JSON数据源写入文件，并可以使用DataFrame阅读器的损坏列功能将损坏的数据写入单独的位置，并使用另一个阅读器从同一作业中写入有效数据。（数据写成拼花）

但是在火花结构流中，我首先通过Kafka作为字符串读取流，然后使用from_json来获取我的数据帧。然后from_json使用JsonToSTRts，它在解析器中使用FailFast模式，并且不会将未解析的字符串返回到DataFrame中的列。（请参阅参考文献中的注释）然后我如何使用SSS？

最后，在批处理作业中，同一作业可以写入两个数据帧。但是Spark结构化流需要对多个接收器进行特殊处理。然后在Spark 2.3中。1（我的当前版本）我们应该包括关于如何正确写入损坏和无效流的详细信息。。。

裁判：https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-Expression-JsonToStructs.html

val rawKafkaDataFrame=spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", config.broker)
  .option("kafka.ssl.truststore.location", path.toString)
  .option("kafka.ssl.truststore.password", config.pass)
  .option("kafka.ssl.truststore.type", "JKS")
  .option("kafka.security.protocol", "SSL")
  .option("subscribe", config.topic)
  .option("startingOffsets", "earliest")

  .load()

val jsonDataFrame = rawKafkaDataFrame.select(col("value").cast("string"))

// does not provide a corrupt column or way to work with corrupt
jsonDataFrame.select(from_json(col("value"), schema)).select("jsontostructs(value).*")

Answer 1

我只是想找出结构化流媒体的_corrupt_记录等价物。这是我想到的；希望它能让你更接近你想要的：

// add a status column to partition our output by
// optional: only keep the unparsed json if it was corrupt
// writes up to 2 subdirs: 'out.par/status=OK' and 'out.par/status=CORRUPT'
// additional status codes for validation of nested fields could be added in similar fashion

df.withColumn("struct", from_json($"value", schema))
  .withColumn("status", when($"struct".isNull, lit("CORRUPT")).otherwise(lit("OK")))
  .withColumn("value", when($"status" <=> lit("CORRUPT"), $"value"))
  .write
  .partitionBy("status")
  .parquet("out.par")

Answer 2

当您从string转换为json时，如果它不能用提供的架构进行解析，它将返回null。您可以筛选空值并选择字符串。像这样的东西。

val jsonDF =  jsonDataFrame.withColumn("json", from_json(col("value"), schema))
val invalidJsonDF = jsonDF.filter(col("json").isNull).select("value")