使用Spark和Scala扁平化json文件

Question

我有一个这样的json文件：

{    
  "Item Version" : 1.0,    
  "Item Creation Time" : "2019-04-14 14:15:09",        
  "Trade Dictionary" : {    
    "Country" : "India",    
    "TradeNumber" : "1",    
    "action" : {    
      "Action1" : false,    
      "Action2" : true,    
      "Action3" : false    
    },    
    "Value" : "XXXXXXXXXXXXXXX",    
    "TradeRegion" : "Global"    
  },    
  "Prod" : {    
    "Type" : "Driver",    
    "Product Dic" : { },    
    "FX Legs" : [ {    
      "Spot Date" : "2019-04-16",        
      "Value" : true    
    } ]    
  },    
  "Payments" : {    
    "Payment Details" : [ {    
      "Payment Date" : "2019-04-11",    
      "Payment Type" : "Rej"
    } ]
  }
}

我需要以下格式的表格：

Version|Item Creation Time|Country|TradeNumber|Action1|Action2|Action3|Value |TradeRegion|Type|Product Dic|Spot Date |Value|Payment Date|Payment Type |
1 |2019-04-14 14:15 | India| 1 | false| true | false |xxxxxx|Global |Driver|{} |2019-04-16 |True |2019-11-14 |Rej

所以它只需迭代每个键值对，将键作为列名，并将其值放入表值。

我当前的代码：

val data2 = data.withColumn("vars",explode(array($"Product")))
  .withColumn("subs", explode($"vars.FX Legs"))
  .select($"vars.*",$"subs.*")

这里的问题是我必须自己提供列名。有什么方法可以让这个更通用吗？

Answer 1

这个解决方案可以很容易地使用一个名为JFlot-https://github.com/opendevl/Json2Flat.的库来实现

String str = new String(Files.readAllBytes(Paths.get("/path/to/source/file.json")));

JFlat flatMe = new JFlat(str);

//get the 2D representation of JSON document
List<Object[]> json2csv = flatMe.json2Sheet().getJsonAsSheet();

//write the 2D representation in csv format
flatMe.write2csv("/path/to/destination/file.json");

Answer 2

使用分解函数用数组展平数据帧。下面是一个示例：

val df = spark.read.json(Seq(json).toDS.rdd)
df.show(10, false)
df.printSchema

df: org.apache.spark.sql.DataFrame = [Item Creation Time: string, Item Version: double ... 3 more fields]
+-------------------+------------+--------------------------------+----------------------------------------+---------------------------------------------------+
|Item Creation Time |Item Version|Payments                        |Prod                                    |Trade Dictionary                                   |
+-------------------+------------+--------------------------------+----------------------------------------+---------------------------------------------------+
|2019-04-14 14:15:09|1.0         |[WrappedArray([2019-04-11,Rej])]|[WrappedArray([2019-04-16,true]),Driver]|[India,1,Global,XXXXXXXXXXXXXXX,[false,true,false]]|
+-------------------+------------+--------------------------------+----------------------------------------+---------------------------------------------------+
root
 |-- Item Creation Time: string (nullable = true)
 |-- Item Version: double (nullable = true)
 |-- Payments: struct (nullable = true)
 |    |-- Payment Details: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- Payment Date: string (nullable = true)
 |    |    |    |-- Payment Type: string (nullable = true)
 |-- Prod: struct (nullable = true)
 |    |-- FX Legs: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- Spot Date: string (nullable = true)
 |    |    |    |-- Value: boolean (nullable = true)
 |    |-- Type: string (nullable = true)
 |-- Trade Dictionary: struct (nullable = true)
 |    |-- Country: string (nullable = true)
 |    |-- TradeNumber: string (nullable = true)
 |    |-- TradeRegion: string (nullable = true)
 |    |-- Value: string (nullable = true)
 |    |-- action: struct (nullable = true)
 |    |    |-- Action1: boolean (nullable = true)
 |    |    |-- Action2: boolean (nullable = true)
 |    |    |-- Action3: boolean (nullable = true)


val flat = df
    .select($"Item Creation Time", $"Item Version", explode($"Payments.Payment Details") as "row")
    .select($"Item Creation Time", $"Item Version", $"row.*")
flat.show

flat: org.apache.spark.sql.DataFrame = [Item Creation Time: string, Item Version: double ... 2 more fields]
+-------------------+------------+------------+------------+
| Item Creation Time|Item Version|Payment Date|Payment Type|
+-------------------+------------+------------+------------+
|2019-04-14 14:15:09|         1.0|  2019-04-11|         Rej|
+-------------------+------------+------------+------------+

Answer 3

由于数组列和结构列在多个级别混合在一起，因此创建通用解决方案并不是那么简单。主要问题是，必须在作为动作的所有数组列上执行分解函数。

我能想到的最简单的解决方案使用递归来检查任何结构或数组列。如果有，那么这些将被展平，然后我们再次检查（展平后会有额外的列，可以是数组或结构，因此很复杂）。flattenstruct部分来自这里。

代码：

def flattenStruct(schema: StructType, prefix: String = null) : Array[Column] = {
  schema.fields.flatMap(f => {
    val colName = if (prefix == null) f.name else (prefix + "." + f.name)   
    f.dataType match {
      case st: StructType => flattenStruct(st, colName)
      case _ => Array(col(colName))
    }
  })
}

def flattenSchema(df: DataFrame): DataFrame = {
    val structExists = df.schema.fields.filter(_.dataType.typeName == "struct").size > 0
    val arrayCols = df.schema.fields.filter(_.dataType.typeName == "array").map(_.name)

    if(structExists){
        flattenSchema(df.select(flattenStruct(df.schema):_*))
    } else if(arrayCols.size > 0) {
        val newDF = arrayCols.foldLeft(df){
          (tempDf, colName) => tempDf.withColumn(colName, explode(col(colName)))
        }
        flattenSchema(newDF)
    } else {
        df
    }
}

在输入数据帧上运行上述方法：

flattenSchema(data)

将给出一个具有以下模式的数据框：

root
 |-- Item Creation Time: string (nullable = true)
 |-- Item Version: double (nullable = true)
 |-- Payment Date: string (nullable = true)
 |-- Payment Type: string (nullable = true)
 |-- Spot Date: string (nullable = true)
 |-- Value: boolean (nullable = true)
 |-- Product Dic: string (nullable = true)
 |-- Type: string (nullable = true)
 |-- Country: string (nullable = true)
 |-- TradeNumber: string (nullable = true)
 |-- TradeRegion: string (nullable = true)
 |-- Value: string (nullable = true)
 |-- Action1: boolean (nullable = true)
 |-- Action2: boolean (nullable = true)
 |-- Action3: boolean (nullable = true)

要将结构列的前缀保留在新列的名称中，只需在flattstruct函数中调整最后一个大小写：

case _ => Array(col(colName).as(colName.replace(".", "_")))