是否可以在 Pyspark 中对 DataFrame 进行子类化？

Posted 2023-04-15

【中文标题】是否可以在 Pyspark 中对 DataFrame 进行子类化？

【英文标题】：Is it possible to subclass DataFrame in Pyspark?

【发布时间】：2017-01-11 18:43:24

【问题描述】：

Pyspark 的文档显示 DataFrame 是由 sqlContext、sqlContext.read() 和各种其他方法构建的。

（见https://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html）

是否可以继承 Dataframe 并独立实例化它？我想为基础 DataFrame 类添加方法和功能。

【参考方案1】：

这真的取决于你的目标。

从技术上讲，这是可能的。 pyspark.sql.DataFrame 只是一个普通的 Python 类。如果需要，您可以对其进行扩展或猴子补丁。

from pyspark.sql import DataFrame

class DataFrameWithZipWithIndex(DataFrame):
     def __init__(self, df):
         super(self.__class__, self).__init__(df._jdf, df.sql_ctx)

     def zipWithIndex(self):
         return (self.rdd
             .zipWithIndex()
             .map(lambda row: (row[1], ) + row[0])
             .toDF(["_idx"] + self.columns))

示例用法：

df = sc.parallelize([("a", 1)]).toDF(["foo", "bar"])

with_zipwithindex = DataFrameWithZipWithIndex(df)

isinstance(with_zipwithindex, DataFrame)

True

with_zipwithindex.zipWithIndex().show()

+----+---+---+
|_idx|foo|bar|
+----+---+---+
|   0|  a|  1|
+----+---+---+

实际上，您将无法在这里做很多事情。 DataFrame 是 JVM 对象的一个​​瘦包装器，除了提供文档字符串、将参数转换为本机所需的形式、调用 JVM 方法以及在必要时使用 Python 适配器包装结果之外，并没有做太多其他事情。

使用纯 Python 代码，您甚至无法接近 DataFrame / Dataset 内部或修改其核心行为。如果您正在寻找独立的、仅 Python 的 Spark DataFrame 实现，这是不可能的。