主页 / coding / 问题 / 77252446
Accepted
user2153235
Asked: 2023-10-08 13:41:45 +0800 CST

将 CSV 读入 Spark DataFrame 的两种方法有何不同？

  • 772

通过遵循此Spark 示例 教程 （以及同一站点上的其他教程），我逐渐熟悉了 Python、Spark 和 PySpark 的基础知识。一开始，他们提供了三种读取同一文件的方法：

spark.read.csv("/tmp/resources/zipcodes.csv")
spark.read.format("csv") \
                  .load("/tmp/resources/zipcodes.csv")
spark.read.format("org.apache.spark.sql.csv") \
                  .load("/tmp/resources/zipcodes.csv")

这里，spark是 类 的一个对象 pyspark.sql.session.SparkSession。该课程指出，第二个和第三个命令是第一个命令的替代命令，但用于“完全限定的数据源名称”。不幸的是，PySpark 中的文档字符串非常简洁。然而，所有三个示例都使用了完全限定路径，因此命令的解释spark.read.format 似乎非常不完整。

方法调用之间有什么区别？对我来说，需要一个全新的专用csv方法来专门处理 CSV，这似乎很奇怪——除非它只是format 具有 CSV 特定便利性的方法的包装器。

我发现了什么

我发现的一个内容丰富的页面是SaturnCloud 页面，但我对该format方法更通用且速度更慢的解释感到困惑。如果该方法是包装器，我看不到这种情况csv- 除非摄取器是以高度次优的方式设置的，并且在每个记录、每个字段或每个字符的基础上有大量控制流。

同一站点还将该csv方法称为 format("csv"). 这表明它甚至不提供包装器可能提供的任何附加功能，并且它根本不应该变慢。所以这个网站是自相矛盾的。

此页面 将该csv方法称为 的“快捷方式” format("csv")。同样，这给人一种感觉，它是一个薄包装器，但这与 SaturnCloud 的指示（可能存在性能差异）不一致，也不符合Spark By Examples的暗示（它们适用于不同形式的数据源名称）。

有关差异的问题之前已作为 Stack Overflow 评论提出 。

csv

1 个回答

  1. Best Answer

    让我们通过源码来揭开这个谜团吧！我假设您使用的是 Spark v3.5.0，这是撰写本文时的最新版本。

    如果我们看一下DataFrameReader.scala的csv方法，我们会看到以下内容：

    @scala.annotation.varargs
def csv(paths: String*): DataFrame = format("csv").load(paths : _*)

    这告诉我们，确实，做spark.read.csv()和做的作用spark.read.format("csv").load()是完全一样的。性能上应该没有区别。

    又怎样呢spark.read.format("org.apache.spark.sql.csv").load()？

    我以前从未见过这个，所以我决定在 pyspark shell 中尝试一下：

    >>> df = spark.read.format("org.apache.spark.sql.csv").load("random_file.csv")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
    ...
        at org.apache.spark.sql.errors.QueryExecutionErrors$.failedToFindDataSourceError(QueryExecutionErrors.scala:587)
        at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:675)
    ...
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.csv.DefaultSource
    ...

    这不起作用，我得到一个ClassNotFoundException.

    经过一番挖掘，我在源代码中找到了这个 Map ，它本质上将完全限定的数据源名称映射到它们的简写（到目前为止我们一直在这样做）。最重要的一行是这一行：

    "com.databricks.spark.csv" -> csv,

    然后我在 Spark shell 中尝试了：

    >>> df1 = spark.read.format("com.databricks.spark.csv").load("random_file.csv")

    那行得通！

    结论

    • spark.read.csv()并spark.read.format("csv").load()做完全相同的事情，前者是后者的一层非常薄的包装纸
    • org.apache.spark.sql.csv不是 CSV 文件的正确完全限定数据源名称：它是com.databricks.spark.csv
    • 2

相关问题