Scalaschema_of_json函式在spark結構化流中失敗

我創建了一個函式來讀取 JSON 作為帶有其模式的字串。然后在火花流中使用該功能。這樣做時我遇到了錯誤。當我首先創建模式，然后使用該模式進行讀取時，同樣的部分作業，但不能在單行中作業。我該如何解決？

def processBatch(microBatchOutputDF: DataFrame, batchId: Long) {
  
  TOPICS.split(',').foreach(topic =>{
    var TableName = topic.split('.').last.toUpperCase
    var df = microBatchOutputDF
    
    /*var schema = schema_of_json(df
                                .select($"value")
                                .filter($"topic".contains(topic))
                                .as[String]
                               )*/
    
    var jsonDataDf = df.filter($"topic".contains(topic))
                      .withColumn("jsonData", from_json($"value", schema_of_json(lit($"value".as[String])), scala.collection.immutable.Map[String, String]().asJava))

    var srcTable = jsonDataDf
                    .select(col(s"jsonData.payload.after.*"), $"offset", $"timestamp")

    srcTable
      .select(srcTable.columns.map(c => col(c).cast(StringType)) : _*)
      .write
      .mode("append").format("delta").save("/mnt/datalake/raw/kafka/"   TableName)
    
    spark.sql(s"""CREATE TABLE IF NOT EXISTS kafka_raw.$TableName USING delta LOCATION '/mnt/datalake/raw/kafka/$TableName'""")
  } )
}

火花流代碼

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

val StreamingQuery = InputDf
                        .select("*")
                        .writeStream.outputMode("update")
                        .option("queryName", "StreamingQuery")
                        .foreachBatch(processBatch _)
                        .start()

錯誤： org.apache.spark.sql.AnalysisException：架構應該以 DDL 格式指定為字串文字或 schema_of_json/schema_of_csv 函式的輸出，而不是 schema_of_json(value)

uj5u.com熱心網友回復：

錯誤 –org.apache.spark.sql.AnalysisException：架構應該以 DDL 格式指定為字串文字或 schema_of_json/schema_of_csv 函式的輸出，而不是 schema_of_json(value)

上述錯誤表明from_json()功能存在問題。

語法：- from_json(jsonStr, schema[, options])- 回傳具有給定jsonStrand的結構值schema。

請參閱以下示例：

> SELECT from_json('{"a":1, "b":0.8}', 'a INT, b DOUBLE');
 {"a":1,"b":0.8}
> SELECT from_json('{"time":"26/08/2015"}', 'time Timestamp', map('timestampFormat', 'dd/MM/yyyy'));
 {"time":2015-08-26 00:00:00}

參考 - https://docs.databricks.com/sql/language-manual/functions/from_json.html

uj5u.com熱心網友回復：

這就是我解決這個問題的方法。我從 kafka 輸出資料幀創建了一個過濾資料幀，并像以前一樣應用了其中的所有邏輯。讀取時生成模式的問題是，from_json不知道要從資料幀的所有行中使用哪一行。

def processBatch(microBatchOutputDF: DataFrame, batchId: Long) {
  
  TOPICS.split(',').foreach(topic =>{
    var TableName = topic.split('.').last.toUpperCase
    var df = microBatchOutputDF.where(col("topic") === topic)
    
    var schema = schema_of_json(df
                                .select($"value")
                                .filter($"topic".contains(topic))
                                .as[String]
                               )
    
    var jsonDataDf = df.withColumn("jsonData", from_json($"value", schema, scala.collection.immutable.Map[String, String]().asJava))

    var srcTable = jsonDataDf
                    .select(col(s"jsonData.payload.after.*"), $"offset", $"timestamp")

    srcTable
      .select(srcTable.columns.map(c => col(c).cast(StringType)) : _*)
      .write
      .mode("append").format("delta").save("/mnt/datalake/raw/kafka/"   TableName)
    
    spark.sql(s"""CREATE TABLE IF NOT EXISTS kafka_raw.$TableName USING delta LOCATION '/mnt/datalake/raw/kafka/$TableName'""")
  } )
}

標籤：json 斯卡拉 数据块 阿帕奇卡夫卡流 天蓝色数据块

上一篇：錯誤的“檢測到組合回圈”

下一篇：Scala3：“斷言失敗：表示類Int在運行1中無效。”