我可以创建一个以数据框为元素的数据框吗？ （使用 sqlContext 的 Pyspark）

Posted 2023-04-15

【中文标题】我可以创建一个以数据框为元素的数据框吗？ （使用 sqlContext 的 Pyspark）

【英文标题】：Can I create a data frame that has data frame as its elements? (Pyspark using sqlContext)

【发布时间】：2017-09-04 19:48:43

【问题描述】：

我有 15k 个小文件。处理后，每个文件生成三个Data frames对象：v、vT和e。我想将所有 15k 文件的所有这三个数据帧存储到一个数据帧对象（我们称之为组合）并写入磁盘，以便下次我只读取一次而不是 15k 次。此外，我还可以将此组合转换为 RDD，并一次将 map 函数应用于 15k 条记录。这将充分利用 CPU。

但目前我的实现是将 v、vT 和 e 写入一个文件夹（每个文件夹每个文件。总共 15k 个文件夹）。在每个文件夹中，分别有 v、vT 和 e 三个文件夹。现在我必须读取所有文件的每 15k 次（从技术上讲，我需要读取 15k * 3 = 45k 次）。当我对这些文件应用某些算法时，我只需使用 for 循环将算法一一应用到它们上。我知道不是很聪明

所以我想出了一个想法，即将 15k 个文件中的 v、vT 和 e 存储到一个名为 combo 的列表中，然后创建一个单独的数据框 combo_df。通过将 combo_df 转换为 RDD，我可以使用 map 函数一次在所有 15k 上应用用户定义的函数。

代码如下，针对每个文件：

v = sqlContext.createDataFrame(uri, 
                                ['id', 'URI', 'flag'])
vT = sqlContext.createDataFrame(vertex, 
                               ['id', 'URI_ID_FK', 'Vertex_Type_URI_ID_FK'])
e = sqlContext.createDataFrame(edge, 
                                ['src', 'dst', 'Relation_Type_URI_ID_FK'])

uri、vertex 和 edge 是从每个文件中提取的三个列表对象。

将 15k 文件中的所有 v、vT 和 e 存储到单个列表组合中

combo = [[v1, vT2, e3],...,[vN, vTN, eN]] (pseudo-code)

我想使用组合创建一个数据框（combo_df）

combo_df = sqlContext.createDataFrame(combo, ['v', 'vT', 'e'])

此时出现错误：

AssertionError: dataType 应该是 DataType

我不知道如何解决这个问题。

【参考方案1】：

我可以创建一个以数据框为元素的数据框吗？

你不能。只需分别写每个DataFrame。

【讨论】：

我有 15k 个文件。在这种情况下，我将不得不创建 45k 个文件。这些文件非常小，例如 100kb 到 300kb。所以我需要找到一种方法将它们组合成一个数据帧，这意味着给我 15k 文件，我给你一个数据帧。

【参考方案2】：

在我看来，您希望将三个数据框分开，因为它们的结构。与其通过读取每个文件来创建三个不同的数据框，不如考虑只保留一个并将列重新组织为structType() 以用于嵌套数据框。你最终会得到这样的模式：

    root
     |-- v: struct (nullable = false)
     |    |-- id: string (nullable = true)
     |    |-- URI: string (nullable = true)
     |    |-- flag: string (nullable = true)
     |-- vT: struct (nullable = false)
     |    |-- id: string (nullable = true)
     |    |-- URI_ID_FK: string (nullable = true)
     |    |-- Vertex_Type_URI_ID_FK: string (nullable = true)
     |-- e: struct (nullable = false)
     |    |-- src: string (nullable = true)
     |    |-- dst: string (nullable = true)
     |    |-- Relation_Type_URI_ID_FK: string (nullable = true)

它的功能是

from pyspark.sql.functions import struct

【讨论】：

非常感谢。这可能是一个解决方案。我遇到的另一个问题是如何将所有这些 15k 文件的数据帧组合在一起并立即对其应用地图功能。我想这样做的原因是，现在我基本上使用了一个 for 循环来循环它们。在这种情况下，CPU 使用率非常低。令人惊讶的是，Spark 客户端模式（驱动程序并在单个处理器上工作）比集群模式（驱动程序一个，工作人员两个）快得多。我很困惑。

如果您在循环中迭代，您将失去 spark 的分布式方面。您可以通过将父目录指定为路径而不是文件路径来读取一个设置中的所有文件。