Spark中将csv文件加载到RDD和Dataframe中的区别答案

【问题标题】：Difference between loading a csv file into RDD and Dataframe in sparkSpark中将csv文件加载到RDD和Dataframe中的区别
【发布时间】：2018-11-29 09:31:44
【问题描述】：

我不确定这个特定问题是否被提前提出。可能是重复的，但我找不到持续存在的用例。

据我们所知，我们可以将 csv 文件直接加载到数据帧中，也可以将其加载到 RDD 中，然后再将该 RDD 转换为数据帧。

RDD = sc.textFile("pathlocation")

我们可以在这个RDD上应用一些Map、filter等操作，并可以将其转化为dataframe。

我们也可以创建一个直接读取 csv 文件的数据框

Dataframe = spark.read.format("csv").schema(schema).option("header","false").load("pathlocation")

我的问题是，当我们必须首先使用 RDD 加载文件并将其转换为数据帧时，可能会出现什么用例？

我只知道 textFile 逐行读取数据。当我们必须选择 RDD 方法而不是数据帧时，可能会出现什么情况？

【问题讨论】：

标签： csv apache-spark-sql rdd

【解决方案1】：

由于 2 个强大的功能，DataFrames / Datasets 比 RDD 提供了巨大的性能改进：

自定义内存管理（又名 Project Tungsten） 数据以二进制格式存储在堆外内存中。这节省了大量的内存空间。也没有涉及垃圾收集开销。通过提前了解数据的模式并以二进制格式高效存储，也避免了昂贵的 java 序列化。
优化的执行计划（又名 Catalyst Optimizer）
使用 Spark 催化剂优化器创建查询计划以供执行。在经过一些步骤准备好优化的执行计划后，最终执行仅在 RDD 内部发生，但对用户完全隐藏。

一般来说，除非您想自己处理低级优化/序列化，否则您不应该使用 RDD。

PySpark 中的客户分区器实现，带有 RDD：

def partitionFunc(key):
import random
if key == 17850 or key == 12583:
return 0
else:
return random.randint(1,2)

# You can call the Partitioner as below:
keyedRDD = rdd.keyBy(lambda row: row[6])
keyedRDD\
.partitionBy(3, partitionFunc)\
.map(lambda x: x[0])\
.glom()\
.map(lambda x: len(set(x)))\
.take(5)

【讨论】：

什么样的低级优化？你能详细说明一下吗？
1.自定义分区 - 使整个集群中的数据分布均匀，以便您可以解决像倾斜这样的问题。结构化 API（DataFrame/Dataset）中不提供自定义分区器。
2.实现自定义序列化。
是的，这是一个很好的案例。感谢您指出。但是您为编写自定义分区器而继承的类是用 Scala 编写的。我可以使用 scala 内置类在我的 Pyspark 代码中创建自定义分区器.. ??是否可以在 Pyspark 应用程序中创建自定义分区器..？我只是问。我还没有探索它。
谢谢。是的，在 pyspark 中实现自定义分区器非常容易。在上述答案中提供了 PySpark 中 Customer Partitioner 的示例实现。

【解决方案2】：

通常不建议将 RDD 转换为 DF，除非没有 API 可以直接将数据加载为 Dataframe。

This 和this 是两个详细回答您问题的博客。引用前者，

什么时候使用 RDD？在以下情况下考虑使用 RDD 的这些场景或常见用例：

您希望对数据集进行低级转换、操作和控制；

您的数据是非结构化的，例如媒体流或文本流；

您希望使用函数式编程结构而不是特定领域的表达式来操作您的数据；

您不关心在按名称或列处理或访问数据属性时强加模式，例如列格式；

您可以放弃一些优化和性能优势，DataFrames 和 Datasets 可用于结构化和半结构化数据。

【讨论】：