通用方法写

  1. df1.write.format("json、csv、...").mode("overwrite").save("output/persons2")

Json读写

/
注意：默认一行只读取一个json
设置读取的参数： DataFrameReader.385行
lineSep： 每行的分隔符，默认 \r, \r\n and \n
mode（写股模式）:
append ： 追加写
ignore : 输出目录存在，就不写
error（默认） : 输出目录存在，就报错
overwrite : 覆盖写
/

  2.     @Test
  3.       def testReadAndWriteJson():Unit={
  5.          // 专门读json
  6.          val dataFrame1 = session.read.json("input/people.json")
  9.       // 通用的读取方法   通过format(文件格式)不同，读取不同格式的数据
  10.          val dataFrame2 = session.read
  11.            .format("json")
  12.            .option("lineSep"," ")   //设置读取的参数
  13.            .load("input/people.json")
  15.       dataFrame2.show()
  18.       // 写JSON
  19.       var datas:java.util.List[Row]=new util.ArrayList[Row]();
  21.       datas.add(Row("jack",20))
  22.       datas.add(Row("jack1",30))
  23.       datas.add(Row("jack2",40))
  26.       // 表结构
  27.       var schema: StructType=StructType(  StructField("name",StringType) :: StructField("age",IntegerType) :: Nil)
  29.       val df: DataFrame = session.createDataFrame(datas, schema)
  31.       val rdd: RDD[Row] = df.rdd.coalesce(1)
  33.      // rdd.toDS()
  36.       val df1: DataFrame = session.createDataFrame(rdd, schema)
  38.       // 通用方法写json
  39.       df1.write.format("json").mode("overwrite").save("output/persons2")
  41.       // 专门写json格式
  42.       df.write.mode("ignore").json("output/persons1.json")
  45.       }

csv读写

/
.csv 是一种特殊的文件格式，在这种格式中保存的数据，必须以一种指定的分隔符分割
c(comma)s(seperate)v(value): 逗号分割字段
:sv :割字段
asv : a符号分割字段
注意： 默认只认,分隔符
设置读取的参数： DataFrameReader.608行
sep (default ,)
header (default false)
/

  2.     @Test
  3.       def testCSV():Unit={
  5.       val df: DataFrame = session.read.option("sep", ";").option("header", "true").csv("input/people.csv")
  7.       df.write
  8.         .option("sep",",")
  9.         .option("header","true")
  10.         .csv("output/csv")
  11.       }

jdbc读写

/*
def jdbc(
url: String,
table: String,

  1.   columnName: String,  : 只能传入numeric, date, or timestamp，用来分区<br />      lowerBound: Long,   ：  分区列的最小范围<br />      upperBound: Long,   :  分区列的最大范围<br />      numPartitions: Int  ： 分几个区
  3.   connectionProperties: Properties  ： <br /> username,<br />password,<br />driverClass<br />)<br />      参数名参考：  JDBCOptions<br />   */

jdbc读

  2.     @Test
  3.       def testJDBCRead():Unit= {
  5.       val properties = new Properties()
  7.       properties.put("user", "root")
  8.       properties.put("password", "000000")
  9.       properties.put("driverClass", "com.mysql.jdbc.Driver")
  11.       val df: DataFrame = session.read.jdbc("jdbc:mysql://hadoop104:3306/gmall_report",
  12.         "ads_user_total",
  13.         properties
  14.       )
  16.       //全表显示，默认显示前20行
  17.       df.show()
  19.       // 按需查询
  20.       df.createTempView("ads_user_total")
  22.       session.sql("select dt,no_order_user_count from ads_user_total where dt >= '2021-05-08' ").show()
  23.     }

jdbc写

  1. @Test
  2.       def testJDBCWrite():Unit={
  4.       val df: DataFrame = session.read.option("sep", ";").option("header", "true").csv("input/people.csv")
  6.       val properties = new Properties()
  8.       properties.put("user", "root")
  9.       properties.put("password", "000000")
  10.       properties.put("driverClass", "com.mysql.jdbc.Driver")
  12.       // 写入jdbc的数据是默认text格式，如果要用idea导数据最好先在mysql中创建好表。
  13.       df.write.mode("overwrite").jdbc("jdbc:mysql://hadoop104:3306/gmall_report","people",properties)
  14.       }

不指定输出格式默认parquet

/
不指定输出文件的格式(json\csv等)，默认以parquet格式写 ，默认读也是读parquet文件
Avro SequnceFile ： 早期
ORC（Hive研发）： spark2.4.x以上才支持orc
Parquet(Hadoop生态圈通用) : phoenix,impala,presto
textfile : 一直使用，简单，方便查看
textfile + 压缩格式
/

  2.     @Test
  3.       def testDefault():Unit={
  5.       val df: DataFrame = session.read.option("sep", ";").option("header", "true").csv("datas/people.csv")
  7. //      df.write.save("output/default")
  9.       session.read.load("datas/users.parquet").show()
  11.       }