需求
讀取HDFS中CSV文件的指定列，並對列進行重命名，並保存回HDFS中
原數據展示
movies.csv
操作后數據展示
注：
write.format()支持輸出的格式有 JSON、parquet、JDBC、orc、csv、text等文件格式
save()定義保存的位置，當我們保存成功后可以在保存位置的目錄下看到文件，但是這個文件並不是一個文件而是一個目錄。
不用擔心，這是沒錯的，我們讀取的時候，
並不需要使用文件夾里面的part-xxxx文件，直接讀取目錄即可。
代碼
# -*- coding: utf-8 -*-
from pyspark import SparkContext
from pyspark.sql import SparkSession
import json
import pandas as pd
''' 當需要把Spark DataFrame轉換成Pandas DataFrame時，可以調用toPandas()； 當需要從Pandas DataFrame創建Spark DataFrame時，可以采用createDataFrame(pandas_df)。 但是，需要注意的是，在調用這些操作之前， 需要首先把Spark的參數spark.sql.execution.arrow.enabled設置為true， 因為這個參數在默認情況下是false '''
# 所需字段和新老字段映射關系
columns_json_str = '{"name":"影片名稱","box_office":"票房"}'
columns_dict = json.loads(columns_json_str)
# 獲取spark的上下文
sc = SparkContext('local', 'spark_file_conversion')
sc.setLogLevel('WARN')
spark = SparkSession.builder.getOrCreate()
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
# 讀取本地或HDFS上的文件【.load('hdfs://192.168.3.9:8020/input/movies.csv')】
df = spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('hdfs://192.168.3.9:8020/input/movies.csv')
print(df.dtypes)
# 將spark.dataFrame轉為pandas.DataFrame，在此處選取指定的列
df = pd.DataFrame(df.toPandas(),columns=columns_dict.keys())
print(df)
data_values=df.values.tolist()
data_coulumns=list(df.columns)
#將pandas.DataFrame轉為spark.dataFrame，需要轉數據和列名
df = spark.createDataFrame(data_values,data_coulumns)
# 字段重命名
# df = df.withColumnRenamed('name', '影片名稱')
for key in columns_dict.keys() :
df = df.withColumnRenamed(key , columns_dict[key]);
print(df.collect())
print(df.printSchema())
# 將重命名之后的數據寫入到文件
filepath = 'new_movies.csv'
df.write.format("csv").options(header='true', inferschema='true').save('hdfs://192.168.3.9:8020/input/' + filepath)