pyspark學習系列（二）讀取CSV文件 為RDD或者DataFrame進行數據處理

一、本地csv文件讀取：

最簡單的方法：

import pandas as pd
lines = pd.read_csv(file)
lines_df = sqlContest.createDataFrame(lines)

或者采用spark直接讀為RDD 然后在轉換

import pandas as pd
from pyspark.sql import SparkSession
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *
spark=SparkSession.builder.appName("lz").getOrCreate()
sc = SparkContext.getOrCreate()

lines = sc.textFile('/Users/xdstar/Desktop/test04.csv')
如果你的csv文件有標題 的話，需要剔除首行

header = lines.first()#第一行 print(header)

lines = lines.filter(lambda row:row != header)#刪除第一行

/Users/xdstar/PycharmProjects/untitled3/venv/bin/python /Users/xdstar/Desktop/sprak_test/test1.py
打印結果：
序號,省份,所屬院線,影院編碼,影院名稱,場次編碼,影片編碼,影片名稱,發行版本,影廳編碼,放映日期,放映時間,總人數,總票房

此時lines 為RDD。如果需要轉換成dataframe：

schema = StructType([StructField('HWMC',StringType(),True),StructField('code',StringType(),True)])
lines_df = sqlContest.createDataFrame(lines,schema)
二、hdfs上的csv文件讀取：

1，采用先讀為RDD再轉換的形式

2，采用sqlContext.read.format（），這個有個前提需要提前做好依賴com.databricks.spark.csv

sqlContext = SQLContext(sc)
sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('file')

---------------------
作者：仙人掌_lz
來源：CSDN
原文：https://blog.csdn.net/qq_36603091/article/details/79626675
版權聲明：本文為博主原創文章，轉載請附上博文鏈接！