Spark系列（三十六）SparkSql操作json格式和CSV格式的数据源介绍

在前面的演示案例里面，我们介绍了编写一个简单的spark sql，同时也介绍过读取csv或者json格式很方便，所以这篇文章我们介绍下spark sql读取csv和json格式的数据源。

一、读取json格式

在spark sql中读取json格式的文件非常容易，因为json自带有key键，因此直接读取整个json，进行反序列化即可，所以在编写应用程序的时候，我们只需要填写读取文件的路径即可，例如：

val dataframeRdd = session.read.json("hdfs://43.134.170.235:9000/usersource/user.txt")

二、读取csv格式

在spark sql读取csv格式的数据源的时候会稍微复杂一点，需要定义读取模式，schema等等，先来个案例：

//CSV 格式数据文本文件数据 -> 依据 CSV文件首行是否是列名称，决定读取数据方式不一样的
val dataframeRdd = session.read.format("csv")
  .option("mode","FAILFAST") //读取模式
  .option("sep", "\\t")  //每一行的分隔符
  .option("header", "true")   //文本里面是否包含header头
  .option("inferSchema", "true") // 是否自动推断 schema
  .option("path","hdfs://43.134.170.235:9000/usersource/u.txt") // 文件路径
  .load()

dataframeRdd.show()

在读取csv格式的时候，我们需要定义很多的option参数，下面分别介绍下：

备注：

读取模式，这里我们主要有3种模式，分别是:

PERMISSIVE

当它遇到损坏的记录时，将格式错误的字符串放入由“ columnNameOfCorruptRecord”配置的*字段中，并将其他字段设置为“ null”。为了保留损坏的记录，用户可以在用户定义的模式中设置一个名为columnNameOfCorruptRecord

DROPMALFORMED

忽略整个损坏的记录

FAILFAST

遇到损坏的记录时引发异常

三、写入json格式

在写入的时候比较简单，代码没太多的配置，写入json格式的文件，代码示例如下：

dataframeRdd.where("name == zhangsan").write
  .mode(SaveMode.Overwrite).json("hdfs://43.134.170.235:9000/usersource/u1.json")

四、写入csv格式

dataframeRdd.where("name == zhangsan").write
  .mode(SaveMode.Overwrite).csv("hdfs://43.134.170.235:9000/usersource/u1.csv")

备注：

1、在读取csv格式的时候添加的参数会比较多一点，我们根据实际情况和自身需求进行设置就可以了。

2、在写入的时候不管是哪种文件都提供有对应的API函数，也是直接拿过来开箱即用。