[大数据]pyspark —— spark dataframe 从hdfs读写文件、按照指定文件格式读写文件（读写csv、json、text等文件）、按照指定分隔符读写文件

spark有3种数据结构——RDD、DataFrame、DataSet。这里展示的文件读写方式，都是针对dataFrame数据结构的，也就是文件读进来之后，是一个spark dataFrame。

0、首先连接spark

from pyspark.sql import SparkSession # SparkConf、SparkContext 和 SQLContext 都已经被封装在 SparkSession
spark=SparkSession \
        .builder \
        .appName('my spark task') \
        .getOrCreate()

1、读写hdfs上的文件

SparkSession在读取【hdfs上文件】的时候，可以指定读取文件的格式，支持的文件格式有json, parquet, jdbc, orc, libsvm, csv, text。

1.1 读写hdfs上的文件 ——> 读写csv、json、text文件

【读取】：按照csv文件格式，读取文件

其余的文件格式只需将csv变成相应的文件格式名称即可

from pyspark.sql.types import IntegerType, DoubleType, StringType, StructType, StructField
# 文件在hdfs上的位置
file_path = r"/user/lanyue/data.csv"  
# 指定文件的schema
schema = StructType([
        StructField("column_1", StringType(), True), # nullable=True代表此字段的值不为空
        StructField("column_2", StringType(), True),
        StructField("column_3", StringType(), True),
        ])

# 方法一
# 推荐这种，指定什么文件格式都可以，只需要修改参数format即可
# 不同的格式其load函数会有不同，用的时候请自行搜索。
df = spark.read.format("csv").load(file_path, header=True, inferSchema=True, encoding="utf-8", sep=',') 
# sep=','，表示指定分隔符为逗号，同参数delimiter。
# header=TRUE，表示数据的第一行为列名
# inferSchema，表示是否对字段类型进行推测。=False，默认读取后都按照文本字符处理。=True表示自动推断schema。

# 或者下面这种形式。这两种形式都可以
df = spark.read.format("csv").option("encoding","utf-8").option("header",True).load(file_path, schema=schema)  # 使用指定的schema



# 方法二
df = spark.read.csv(file_path, encoding='utf-8', header=True, inferSchema=True) 
df = spark.read.csv(file_path, encoding='utf-8', header=True, schema=schema) 
# 如果想指定文件格式是json，那就是spark.read.json，其他类似

【保存】：以csv文件的格式保存，注意：是保存在hdfs上，不是保存在本地

其余的文件格式只需将csv变成相应的文件格式名称即可

# 保存在【hdfs上】，以csv文件的格式。指定什么文件格式都可以，只需要修改参数format即可
df.write.mode('append').format("csv").repartition(1).option("encoding","utf-8").option("header",True).save("/lanyue/data.csv") 
# mode=append代表以追加的模式保存，还有其他模式：overwrite覆盖等
# repartition(1), 在yarn模式下，Spark会根据hdfs文件的块数据大小来划分默认的分区数目，但是我们也可以自己设置分区数目，使用参数repartition。=1表示只保存成一个数据块


# 或者
df.write.csv("/lanyue/data.csv", sep="\t", encoding="utf-8", mode='overwrite') # 如果想指定文件格式是json，那就是df.write.json，其他类似

# 如果想保存在【本地】，以csv文件的格式
df = df.toPandas()
df.to_csv(r"data.csv", encoding="utf-8", index=None, header=True)

1.2 读写hdfs上的文件 ——> 按照指定分隔符来读取文件

有的时候，文件中的数据，其分隔符是自定义的，如何指定分隔符来读取文件呢？

可以使用读取csv的格式来读，虽然文件不是csv格式，但是通过spark.read.format(“csv”)的api指定分隔符，从而实现指定分隔符来读取文件的目的。

df = spark.read.format("csv").load(file_path, header=True, inferSchema=True, encoding="utf-8", sep=',') 
# 通过指定参数sep，来指定分隔符，可以是",", "\t"，"\x01"等。同参数delimiter。

1.3、读写hdfs上的文件 —— 读写hive中的表

hive中的表，其本质上，也是保存在hdfs上的文件，其文件格式是orc或者parquet

【读】：从hive表中读取数据

df = spark.sql("select * from hive_table_name")
df.show()

【写】：保存成hive表

# spark-dataframe 保存成hive表
df.write.format("orc").mode("append").partitionBy("dt").saveAsTable(table_name)
# partitionBy，设置分区。如果没有分区，直接去掉该参数即可
# mode="append"，以追加的形式写入

# 直接表to表 
spark.sql = """
	insert overwrite table databases.hive_table_name_2
    partition (saledate) 
    select *  
    from databases.hive_table_name_1
"""

1.4 读写hdfs上的文件 —— 读写mysql中的表

在这里插入代码片