Spark Dataframe 并行读取答案

【问题标题】：Spark Dataframe parallel readSpark Dataframe 并行读取
【发布时间】：2015-05-04 17:57:06
【问题描述】：

使用 pyspark 时，您可以在 sc.textFile 方法中设置减少的数量，以便您可以按照 here 的说明更快地从 S3 读取文件。这很好用，但从 Spark 1.3 开始，我们也可以开始使用 DataFrames。

Spark DataFrames 也可以这样吗？我正在尝试将它们从 S3 加载到 spark 集群（通过 ec2-spark 创建）。基本上我试图让这段代码快速运行非常大的“data.json”文件：

from pyspark import SparkContext
from pyspark.sql import SQLContext
sc = SparkContext(CLUSTER_URL, 'ipython-notebook')
sqlContext = SQLContext(sc)
df = sqlContext.jsonFile('s3n://bucket/data.json').cache()

【问题讨论】：

标签： amazon-s3 apache-spark

【解决方案1】：

实际上有一个与此 here 相关的 TODO 注释，我创建了相应的 issue here，因此如果您需要，您可以对其进行投票。

问候，

奥利维尔。

【讨论】：

【解决方案2】：

在等待问题得到解决的过程中，我找到了一个暂时可行的解决方法。 .json 文件包含每一行的字典，所以我可以首先将其作为 RDD 文本文件读入，然后通过手动指定列将其转换为数据框：

from pyspark import SparkContext
from pyspark.sql import SQLContext, Row
sc = SparkContext(CLUSTER_URL, 'ipython-notebook')
sqlContext = SQLContext(sc)
data = sqlContext.textFile('s3n://bucket/data.json', 30).cache()
df_rdd = data\
    .map(lambda x : dict(eval(x)))\
    .map(lambda x : Row(x1=x['x1'], x2=x['x2'], x3=x['x3'], x4=x['x4']))
df = sqlContext.inferSchema(df_rdd).cache()

根据docs。这也意味着您可以使用.csv 文件而不是json 文件（这通常会节省大量磁盘空间），只要您在spark 中手动指定列名即可。

【讨论】：