Spark scala - Dataframe - 将模式文本文件（输入）转换为复杂的 Json 模式（输出）答案

【问题标题】：Spark scala - Dataframe - Convert Schema Text file (input) to complex Json schema (output)Spark scala - Dataframe - 将模式文本文件（输入）转换为复杂的 Json 模式（输出）
【发布时间】：2018-07-02 14:05:58
【问题描述】：

我有如下输入和输出模式。我是火花和斯卡拉的新手。有人可以帮我转换加载文本文件的数据框，最后将其转换为土地 json 文件。

输入

    val OTBCleanFile: Array[StructField] = Array(
          StructField("varbl_1_txt", StringType, true),
          StructField("varbl_2_txt", StringType, true),
          StructField("varbl_3_txt", StringType, true),
          StructField("varbl_4_txt", StringType, true),
          StructField("varbl_5_txt", StringType, true),
          StructField("varbl_6_txt", StringType, true),
          StructField("varbl_7_txt", StringType, true),
          StructField("varbl_8_txt", StringType, true),
          StructField("varbl_9_txt", StringType, true),
          StructField("varbl_10_txt", StringType, true),
          StructField("varbl_11_txt", StringType, true),
          StructField("varbl_12_txt", StringType, true),
          StructField("varbl_13_txt", StringType, true),
          StructField("varbl_14_txt", StringType, true),
          StructField("varbl_15_txt", StringType, true),
          StructField("email", StringType, true))

输出：

val JsonFileScma = (new StructType)
  .add("col1",  (new StructType)
    .add("col2",  StringType)
    .add("col3",  StringType)
    .add("col4",  StringType)
    .add("col5",  StringType)
    .add("col6",  StringType)
    .add("col7",  StringType))
  .add("email", (new StructType)
    .add("type", StringType)
    .add("value", StringType))
  .add("templateId", StringType)

映射可以是一对一的，并且从输入文件/模式中留下很少的字段。

在此先感谢您，问候，达图

【问题讨论】：

你试过什么？ dataset.map 不工作吗？
读取文件：/* 读取 OTB 文件 iso-8859-1 */ val delimiter2 = "\307" val OTBInputDF = sqlc.read .format("com.databricks.spark.csv") . option("header", "false") // 使用所有文件的第一行作为标题 .option("delimiter", delimiter2) .option("charset","iso-8859-1") .schema(StructType(OTBCleanFile )) .load("OTBCleanFile.dat") OTBInputDF.show() OTBInputDF.printSchema()
val data = OTBInputDF.selectExpr(few fields) data.write .format("json") .save("OTBInputDF.json") 但这并不是我真正想要的输出Json格式

标签： arrays json scala apache-spark spark-dataframe

【解决方案1】：

****工作代码****

    // Initialize

    val conf = new                                                              
    SparkConf().setAppName("process_text_to_json").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlc = new org.apache.spark.sql.SQLContext(sc)    

// Reading File iso-8859-1

val delimiter = "\307"
val OTBInputDF = sqlc.read
  .format("com.databricks.spark.csv")
  .option("header", "false") // Use first line of all files as header
  .option("delimiter", delimiter)
  .option("charset","iso-8859-1")
  .schema(StructType(OTBCleanFile))
  .load("cln.dat")

  // Convert to Json

  val data = OTBInputDF.selectExpr("varbl_1_txt as col1"  ,"varbl_1_txt as col2","varbl_1_txt as col3","varbl_1_txt as col4","varbl_1_txt col5","email as col6")

  val data2 = data.select(to_json(struct(col("col1"),col("col2"),col("col3"),col("col4"),col("col5"))) as "clientTag",col("col6"))
  data2.show()
  data2.printSchema()

【讨论】：