如何可變json欄位名稱火花

我有 json 日志檔案（json delimiter /n）并且需要 spark struct 型別，但是我的 txt 檔案中的每個 json 第一個欄位名稱都不同，我該怎么做？

val elementSchema = new StructType()
.add("name",StringType,true)
.add("object_type",StringType,true)
.add("privilege",StringType,true)

val simpleSchema = new StructType()
.add("authorization_failure",StringType,true)
.add("catalog_objects",elementSchema,true)
.add("impersonator",StringType,true)
.add("network_address",StringType,true)
.add("query_id",StringType,true)
.add("session_id",StringType,true)
.add("sql_statement",StringType,true)
.add("start_time",StringType,true)
.add("statement_type",StringType,true)
.add("status",StringType,true)
.add("user",StringType,true)

val anaSchema = new StructType()
.add("saasd",StringType,true)

val config = new SparkConf()`
config.set("spark.sql.shuffle.partitions","300")

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

val dataframe = spark.read
.json(s"/home/ogn/denemeler/big_data/impala_audit_spark/file/testa.txt")

dataframe.printSchema()

val df =dataframe.select(to_json( struct( dataframe.columns.map(col(`_`)):`_`*  ) ).alias("all"))

期待

每個欄位結構

authorization_failure|catalog_objects|impersonator|network_address|query_id|session_id|sql_statement|start_time|statement_type|status|user|

testa.txt 內容為 單個檔案中有接近 3m 的 json

{"1648039261379":{"query_id":"x","session_id":"da40931781b4b8ed:978bb8edb9177dbd","start_time":"2022-03-23 15:41:01.234826","authorization_failure":false,"status":"","user":"x","impersonator":null,"statement_type":"QUERY","network_address":"x","sql_statement":"y","catalog_objects":[{"name":"_impala_builtins","object_type":"DATABASE","privilege":"VIEW_METADATA"},{"name":"s","object_type":"TABLE","privilege":"SELECT"}]}}
{"1648039261510":{"query_id":"x","session_id":"344247956fada236:7d9c0930b7c51b9a","start_time":"2022-03-23 15:41:01.507023","authorization_failure":false,"status":"","user":"x","impersonator":null,"statement_type":"USE","network_address":"x","sql_statement":"t","catalog_objects":[{"name":"g","object_type":"DATABASE","privilege":"ANY"}]}}

uj5u.com熱心網友回復：

第 1 步：使用textFile將 Json 檔案作為簡單文本檔案讀取：

val ds: Dataset[String] = spark.read.textFile("testa.txt")

第 2 步：使用regexp_extract洗掉第一個 Json 級別。您也可以決議 json 字串，但我認為這種方法更快。

import spark.implicits._
val ds2: Dataset[String] = ds.withColumn("value", regexp_extract('value, "\\{.*:(\\{.*\\})\\}", 1)).as[String]

第 3 步：將字串決議為資料框：

val df3: DataFrame = spark.read.json(ds2)

df3現在有結構

root
 |-- authorization_failure: boolean (nullable = true)
 |-- catalog_objects: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- object_type: string (nullable = true)
 |    |    |-- privilege: string (nullable = true)
 |-- impersonator: string (nullable = true)
 |-- network_address: string (nullable = true)
 |-- query_id: string (nullable = true)
 |-- session_id: string (nullable = true)
 |-- sql_statement: string (nullable = true)
 |-- start_time: string (nullable = true)
 |-- statement_type: string (nullable = true)
 |-- status: string (nullable = true)
 |-- user: string (nullable = true)

標籤：json 斯卡拉 阿帕奇火花 解析

上一篇：是否有可能使用絕對路徑進行htaccess301重定向？

下一篇：如果沒有分配所有資源，則Spark失敗