使用Spark读取CSV

Question

我正在通过Spark使用以下命令读取csv文件。

rdd=sc.textFile("emails.csv").map(lambda line: line.split(","))

我需要创建一个Spark DataFrame。

我使用以下方法将此rdd转换为spark df：

dataframe=rdd.toDF()

但是在将rdd转换为df时，我需要指定df的模式。我试着这样做：（我只有两列文件和消息）

from pyspark import Row

email_schema=Row('file','message')

email_rdd=rdd.map(lambda r: email_schema(*r))

dataframe=sqlContext.createDataFrame(email_rdd)

然而，我得到了一个错误：java。lang.IllegalStateException：输入行没有架构所需的预期值数。需要2个字段，但提供1个值。

我还尝试使用以下方法读取我的csv文件：

rdd=sc.textFile("emails.csv").map(lambda line: line.split(",")).map(lambda line: line(line[0],line[1]))

我收到错误：TypeError：“list”对象不可调用

我尝试使用熊猫将我的csv文件读取到熊猫数据帧中，然后将其转换为触发DataFrame，但我的文件太大了。

我还补充道：

bin/pyspark --packages com.databricks:spark-csv_2.10:1.0.3

并使用以下命令读取我的文件：

df=sqlContext.read.format('com.databricks.spark.csv').options(header='true').load('emails.csv')

我得到了一个错误：java。io。IOException：（startline 1）在封装令牌完成之前达到EOF

我已经浏览了其他几个相关的线程并尝试了如上所述。有人能解释一下我哪里出错了吗？

[在MacOSX上使用Python 2.7、Spark 1.6.2]

已编辑：

前3行如下。我只需要提取电子邮件的内容。我该怎么做？

1 allen-p/_sent_mail/1."Message-ID：

这是我们的预测”

2 allen-p/\u发送邮件/10。“邮件ID：

去参加一个商务会议会给旅行带来乐趣。尤其是当你必须准备演讲的时候。我建议在这里举行商业计划会议，然后在没有任何正式商业会议的情况下旅行。我甚至会尝试就旅行是理想的还是必要的获取一些诚实的意见。

至于商务会议，我认为尝试激发不同群体之间关于什么有效、什么无效的讨论会更有成效。太多时候，演讲者在发言，其他人安静地等待轮到他们。如果以圆桌讨论的形式举行会议可能会更好。

我的建议是去奥斯汀。打高尔夫球，租一艘滑雪艇和摩托艇。去某个地方飞行需要太多时间。“”

3 allen-p/\u sent\u mail/100。“邮件ID：

测试成功。加油！！！“”

Answer 1

如果您有一个大文件，为什么不使用大块的熊猫数据帧，而不是一次加载所有文件，比如：

import pandas as pd
df_pd = pd.read_csv('myfilename.csv',chunksize = 10000)

for i,chunk in enumerate(df1):
    if i==0:
        df_spark = sqlContext.createDataFrame(chunk)
    else:
        df_spark = df_spark.unionAll(sqlContext.createDataFrame(chunk))

df\U spark将是您所需的spark数据帧。这是低效的，但它会起作用。有关实现相同功能的其他方法，可以参考此问题的答案

另一种可能的方法是使用rdd的inferSchema方法，但您需要在csv文件中有列名才能工作，请参阅此。因此，您可以执行以下操作：

srdd = inferSchema(rdd)
email_rdd=rdd.map(lambda r: srdd(*r))

dataframe=sqlContext.createDataFrame(email_rdd)

Answer 2

如果RDD适合内存，则：

rdd.toPandas().to_csv('emails.csv')

如果不是，请将spark csv用于您的spark版本：

rdd.write.format('com.databricks.spark.csv').save('emails.csv')

在您上面的示例中：

rdd=....map(lambda line: line.split(",")).map(lambda line: line(line[0],line[1]))

难道你不想：

rdd=....map(lambda line: line.split(",")).map(lambda line: (line[0], line[1]))