sparksql解析json数据并保存

spark sql DataFrame 自动解析json数据

创建sparksession
val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()

读取json文件，可以自动解析

val df: DataFrame = spark.read.format("json").load(c://a.json)

可以设置DataFrame的字段来查看数据

val frame: DataFrame = df.select("id", "company_name", "eduLevel_name", "emplType", "jobName")
//字段
frame.show()

SPARK – DATAFRAME按指定分隔符读取和写入文件

按指定分隔符读取文件

SparkSession读取时可以指定format，format支持：json, parquet, jdbc, orc, libsvm, csv, text这几种格式

JSON

spark.read.json(inputFile1)
//或者
spark.read.format("json").load(inputFile1)

PARQUET

spark.read.parquet(inputFile1)
//或者
spark.read.format("parquet").load(inputFile1)

JDBC

val connectionProperties = new Properties()
connectionProperties.put("user", "username")
connectionProperties.put("password", "password")
//connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")
spark.read.jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
//或者
spark.read.format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("driver","com.mysql.jdbc.Driver")
      .option("dbtable", "student")
      .option("user", "root")
      .option("password", "hadoop").load()

CSV

csv默认以英文逗号分割

//推断头信息，省略首行的头信息
spark.read.option("inferSchema", "true").option("header", "true").csv(inputFile1)
//或者
spark.read.format("csv").option("inferSchema", "true").option("header", true).load(inputFile1)

TEXT

直接将一行数据作为一列，不做分割

spark.read.text(inputFile1)
//或者
spark.read.format("text").load(inputFile1)

自定义分隔符

有时候我们的输入数据分隔符是自定义的，我们可以使用读取csv的格式来读，虽然文件不是csv格式，但是通过csv的api可以指定分隔符，从而通过指定分隔符来读取数据。csv的option有如下选择：我们可以设置seq参数来设置自定义分隔符

spark.read.option("sep", "\t").csv(inputFile1)
//或者
spark.read.option("delimiter", "\t").csv(inputFile1)

按指定分隔符写入文件

SparkSession写入时可以指定format，format支持：json, parquet, jdbc, orc, libsvm, csv, text这几种格式。

JSON

spark.write.json(inputFile1)
//或者
inputDF.write.format("json").save(outputFile)

PARQUET

spark.write.parquet(inputFile1)
//或者
inputDF.write.format("parquet").save(outputFile)

JDBC

val connectionProperties = new Properties()
connectionProperties.put("user", "username")
connectionProperties.put("password", "password")
//connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")
spark.write
	//.option("createTableColumnTypes", "name CHAR(), comments VARCHAR(1024)")
	.jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
//或者
spark.write.format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("driver","com.mysql.jdbc.Driver")
      .option("dbtable", "student")
      .option("user", "root")
      .option("password", "hadoop").save()

CSV

csv默认以英文逗号分割。

//推断头信息，省略首行的头信息
spark.write.csv(outputFile)
//或者
spark.write.format("csv").save(outputFile)

TEXT

dataframe只能是一列数据，需要提前合并

spark.write.text(outputFile)
//或者
spark.write.format("text").save(outputFile)

// 合并dataframe为一列写入text

val allCols = dataFrame.columns.map(col(_))
dataFrame.select(concat_ws(splitRex, allCols: _*))
  .write.mode(SaveMode.Overwrite).text(saveDir)

自定义分隔符

有时候我们输出数据时需要自定义分隔符，我们仍然可以使用写入csv的格式来写，但是不能通过seq参数来设置了，只能通过delimiter参数设置。

这个重要

inputDF.write.option("delimiter", "\t").csv(outputFile)

eDir)

## 自定义分隔符

  有时候我们输出数据时需要自定义分隔符，我们仍然可以使用写入csv的格式来写，但是不能通过seq参数来设置了，只能通过delimiter参数设置。

这个重要

```scala
inputDF.write.option("delimiter", "\t").csv(outputFile)