SparkSQL数据读写
DataFrameReader
DataFrameWriter
DataFrameReader 对应的组件 SCHEMA OPTION FORMAT
DataFrameReader 有两种访问方式,
一种是使用 load 方法加载, 使用 format 指定加载格式,
还有一种是使用封装方法, 类似 csv, json, jdbc 等
//.第一种形式 READ + FORMAT +load读取
spark.read
.format("csv")
.option("header",value=true)
.option("inferSchema",value = true)
.load("dataset/1231.csv")
//2.第二种形式 使用具体文件类型读取 READ
spark.read
.option("header",value=true)
.option("inferSchema",value = true)
.csv("dataset/1231.csv")
DataFrameWriter 也有两种使用方式, 一种是使用 format 配合 save,
默认的 format 是 Parquet
还有一种是使用封装方法, 例如 csv, json, saveAsTable 等
def parquet(path: String): Unit = { format("parquet").save(path)}
def csv (path: String): Unit = { format("csv").save(path)}
DataFrameWriter
增量操作
使用spark做增量操作的时候,
insertInto 和 mode(SaveMode.Append).saveAsTable()
insertInto
insertInto 使用基于位置
It requires that the schema of the `DataFrame` is the same as the schema of the table.
Unlike `saveAsTable`, `insertInto` ignores the column names and just uses position-based resolution
saveAsTable
基于列名 column_names-based
有不同的模式，
如果是 SaveMode.Overwrite
the schema of the `DataFrame` does not need to be the same as that of the existing table.
如果是 `Append`,
if there is an existing table, we will use the format and options OF the existing table.
The column order in the schema of the `DataFrame` doesn't need to be same as that of the existing table.
Unlike `insertInto`,
`saveAsTable` will use the column names to find the correct column positions
####覆盖
report_info.write.insertInto(tableName="dwd.t_dwd_report_info_d",overwrite=True)
DataFrameWriter 可以将数据保存到 Hive 表中, 所以也可以指定分区和分桶信息
对存储的
SCHEMA
文件的覆盖和追加
Spark的读写
Spark can create distributed datasets from any storage source supported by Hadoop,
including your local file system, HDFS, Cassandra, HBase, Amazon S3,
etc. Spark supports text files, SequenceFiles, and any other Hadoop InputFormat.
写
saveAsTextFile saveAsSequenceFile saveAsSequenceFile
存储问题
存储的载体
存储的数据格式
存储所使用的命令或函数
存数遇到的问题
小文件问题
spark.default.parallelism 在处理RDD时才会起作用，对SparkSql无效。
spark.sql.shuffle.partitions 则是对sparks SQL专用的设置。
源码
org.apache.spark.sql
Interface used to write a [[Dataset]] to external storage systems (e.g. file systems,key-value stores, etc).
Use `Dataset.write` to access this.
使用了 // Builder pattern config options 构建者模式
本地数据写入到Hive表
在Hive中创建表，从外部数据源以txt的形式导入数据
方案一： 利用系统ftp上传数据，并使用系统的数据交换任务将任务写入到Hive表中
方案二： 利用程序 WriteToHive 将数据打包到程序中，程序中读取数据，然后解析写入到Hive表中
01.方案一步骤：
01.查看外部给的数据的情况，确认数据格式正确等情况
做一些必要的查询。修正以及确认工作
02. 创建表
03. 修改数据文件的编码格式以及文件的类型，放到系统的位置-上传到服务器的位置要记录下来
04. 创建或者使用已存在的System_任务，导入数据任务配置和执行
05. 任务执行成功后，确认数据表的情况
02.具体过程
04任务配置和执行
任务流： 数据导出专用
任务 System_数据_ting_no_repeat_f
需要完成的任务：
001.源数据源配置
选择数据源： ftp
基本信息： 文件路径 /test 文件名 test_import_date_info_td.txt
高级配置：
编码 UTF-8
002.目标数据源配置
数据源类型：Hive
数据源名称：
表名： expbase.import_date_info_td
导入配置：
SQL： use expbase; truncate table import_date_info_td ;
Hive文件
行分割符
列分隔符
05.任务执行成功后 -查看数据量以及示例数据情况
select count(*) from expbase.import_date_info_td
select * from expbase.import_date_info_td
03.注意事项
01.数据确认工作：
数据有无重复数据
Excel的数据中的空格，tab的个数，引号等符号问题
异常数据
最终确认数据量，以及示例数据情况
02.创建表的形式
ROW FORMAT DELIMITED FIELDS TERMINATED BY ' '
STORED AS INPUTFORMAT
'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
;
03.修改文件的编码方式和格式，并上传到Unix系统中
编码 UTF-8
文件类型 Unix
04.任务的配置
配置工作
Unix系统的文件 行分割符
数据文件的列分隔符保持一致，在这里是
05.确认工作
确认数据量是否准确
查看示例数据各个字段是否对应，字段的值是否正确
参考
https://github.com/apache/spark/pull/13013
Spark写入hive表时saveAsTable和insertInto的区别 https://blog.csdn.net/huihuixia123/article/details/107658299
Spark（三）-- SparkSQL扩展（数据读写） -- 读写 Parquet、Json 格式文件(二) https://blog.csdn.net/qq_18800463/article/details/101421490
Spark SQL, DataFrames and Datasets Guide http://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html
RDD Programming Guide http://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-programming-guide