如何在加載到Spark數據框之前預處理JSON數據

Question

我想將JSON文件的大目錄轉換爲CSV文件。但是，在將這些JSON文件轉換爲CSV或Dataframe友好格式之前，我需要轉換它們並剪輯它們。這是由transform_json函數完成的。

下面是一個可行的解決方案，但由於json.loads/json.dumps的後面和第四個，它感覺很蠢並且很慢。

rdd = (spark_context.textFile('*.json') 
     .map(json.loads) 
     .flatMap(transform_json) 
     .map(json.dumps)) 

(spark_session.read.json(rdd) 
    .write.format("com.databricks.spark.csv") 
    .option("header", "true") 
    .save("output_dir")) 

我需要把它們放到PySpark Dataframe中，因爲我事先不知道所有的列，Spark會爲我處理。

如何改進此代碼？

Answer 1

看來你想合併模式？ 您可以使用實木複合地板並閱讀： http://spark.apache.org/docs/latest/sql-programming-guide.html#schema-merging