使用pyspark从CSV文件中拆分字段

Question

我在通过PySpark拆分CSV文件时遇到问题。我正在尝试输出葡萄酒的国家和名称（这只是为了证明解析有效），但是出现错误。

这是CSV文件的外观：

，国家，说明，名称，积分，价格，省，region_1，region_2，品种，酒厂

20，US，“ Heitz从1961年开始就用稀有的Grignolino葡萄酿制这种玫瑰红。红宝石葡萄柚红，带有草莓，西瓜，橙皮和咸香料的味道，带有强烈的花香。”，Grignolino 95,24.0，California，Napa Valley，Napa，Rosé，Heitz

这是我的代码

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local").setAppName("SQLProject")
sc = SparkContext(conf = conf)


def parseLine(line):
    fields = line.split(',')
    country = fields[1]
    points = fields[4]
    return country, points


lines = sc.textFile("file:///Users/luisguillermo/IE/Spark/Final Project/wine-reviews/winemag-data-130k-v2.csv")
rdd = lines.map(parseLine)

results = rdd.collect()

for result in results:
    print(result)

并得到此错误：

  File "/Users/luisguillermo/IE/Spark/Final Project/wine-reviews/country_and_points.py", line 10, in parseLine
    points = fields[4]
IndexError: list index out of range

由于描述中有逗号，因此程序似乎很困惑。有关如何解决此问题的任何想法？

Answer 1

我建议使用Spark内置CSV数据源，因为它提供了许多选项，其中包括quotes，该选项用于从列中读取定界符。

quotes

[当您有一列带有用于分隔列的定界符时，请使用引号选项指定引号字符，默认情况下为”，引号内的定界符将被忽略。但是使用此选项，您可以设置任何字符。

如果您想阅读Spark CSV提供的其他选项以及示例，建议阅读以下文章。

spark-read-csv-file-into-dataframe

read-csv

学习愉快！

Answer 2

查看此代码：

df = spark.read\
    .csv('data.csv')

df.printSchema()
df.show()

生成的df是一个DataFrame，其列与CSV一样。查看更多高级功能here