Pyspark读写csv,txt,json,xlsx,xml,avro等文件

1. Spark读写txt文件

读：

df = spark.read.text("/home/test/testTxt.txt").show()
+-------------+
|        value|
+-------------+
|      a,b,c,d|
|123,345,789,5|
|34,45,90,9878|
+-------------+

2. Spark读写csv文件

读：

# 文件在hdfs上的位置
file_path = r"/user/lanyue/data.csv"
# 方法一
# 推荐这种，指定什么文件格式都可以，只需要修改参数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，其他类似

写：

# 保存在【hdfs上】，以csv文件的格式。指定什么文件格式都可以，只需要修改参数format即可
df.repartition(1).write.mode('append').format("csv").option("encoding","utf-8").option("header",True).save("/lanyue/data.csv") 
# mode，保存模式：ovewriter重写、append文件末尾追加、error如果文件存在抛出异常、ignore如果文件存在忽略不更新
# repartition, 在yarn模式下，Spark会根据hdfs文件的块数据大小来划分默认的分区数目，但是我们也可以自己设置分区数目，使用参数repartition。=1表示只保存成一个数据块

# 或者
df.write.csv("/lanyue/data.csv", sep="\t", encoding="utf-8", mode='overwrite') 
# 如果想指定文件格式是json，那就是df.write.json，其他类似
# 通过指定参数sep，来指定分隔符，可以是",", "\t"，"\x01"等。同参数delimiter。

3. Spark读写parquet文件

读：

file = "/user/muzili/data.parquet"
spark_df=spark.read.parquet(file)
df.show()

写：

spark_df.write.parquet(path=file,mode='overwrite')

4. Spark读写json文件

读：

file = "/user/muzili/data.json"
df = spark.read.json(file)
df.show()

写：

df.repartition(1).write.mode('append').format("json").option("encoding","utf-8").option("header",True).save("/user/muzili/data.json")

5. Spark读写excel文件

读：

写：

6. Spark读写xml文件

读：

写：

7. Spark读写orc文件

读：

写：

8. Spark读写avro文件

读：

写：

9. Spark读写mysql中的表

读：

url="jdbc:mysql://host:port/database"
table="table_name"
driver="com.mysql.jdbc.Driver"
user="XXX"
password="XXX"

df = spark.read.format("jdbc")
      .option("url",url) # database地址，格式为jdbc:mysql://主机:端口/数据库
      .option("dbtable",table) # 表名
      .option("user",user)
      .option("password",password)
      .option("driver",driver)
      .load()
      
# 或者以下形式
df = spark.read.format('jdbc').options(url="jdbc:mysql://host:port/database", # database地址
                                     driver="com.mysql.jdbc.Driver",
                                     dbtable="table_name", 
                                     user="XXX",
                                     password="XXX").load()

# 或者以下形式
# mysql的相关配置
prop = {'user': 'xxx', 
        'password': 'xxx', 
        'driver': 'com.mysql.jdbc.Driver'}
url = 'jdbc:mysql://host:port/database' # database地址
df = spark.read.jdbc(url=url, table='mysql_table_name', properties=prop)

写：

# 会自动对齐字段，也就是说，spark_df 的列不一定要全部包含MySQL的表的全部列才行
prop = {'user': 'xxx', 
        'password': 'xxx', 
        'driver': 'com.mysql.jdbc.Driver'}
url = 'jdbc:mysql://host:port/database' # database地址
df.write.jdbc(url=url, table='table_name', mode='append', properties=prop)
# append 追加方式

# 或者以下形式
df.write.format("jdbc")
  .option("url","jdbc:mysql://host:port/database") # database地址
  .option("dbtable","table_name")
  .option("user",user)
  .option("password",password)
  .option("driver",driver)
  .option("batchsize","1000").mode("overwrite") # overwrite 清空表再导入
  .save()