使用 pyspark 将 RDD 行转换为数据帧时出错

Posted 2023-04-15

【中文标题】使用 pyspark 将 RDD 行转换为数据帧时出错

【英文标题】：Error converting RDD row to a dataframe with pyspark

【发布时间】：2018-05-07 03:09:07

【问题描述】：

我编写了一个要应用于数据帧的函数，但首先我必须将数据帧转换为要映射的 RDD。然后我打印出来看看结果：

x = exploded.rdd.map(lambda x: add_final_score(x.toDF()))
print(x.take(2))

函数 add_final_score 需要一个数据帧，这就是为什么我必须在通过之前将 x 转换回 DF。但是，它给了我这个 toDF 不在列表中的错误：

Py4JJavaError                             Traceback (most recent call last)
<ipython-input-491-11e7b77ecf3f> in <module>()
     42 #                      StructField('segmentName', StringType(), True)])
     43 # x = exploded.rdd.map(lambda y: y.toDf())
---> 44 print(x.take(2))

~/spark-2.3.0-bin-hadoop2.7/python/pyspark/rdd.py in take(self, num)
   1356 
   1357             p = range(partsScanned, min(partsScanned + numPartsToTry, totalParts))
-> 1358             res = self.context.runJob(self, takeUpToNumLeft, p)
   1359 
   1360             items += res

~/spark-2.3.0-bin-hadoop2.7/python/pyspark/context.py in runJob(self, rdd, partitionFunc, partitions, allowLocal)
    999         # SparkContext#runJob.
   1000         mappedRDD = rdd.mapPartitions(partitionFunc)
-> 1001         port = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions)
   1002         return list(_load_from_socket(port, mappedRDD._jrdd_deserializer))
   1003 

~/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1158         answer = self.gateway_client.send_command(command)
   1159         return_value = get_return_value(
-> 1160             answer, self.gateway_client, self.target_id, self.name)
   1161 
   1162         for temp_arg in temp_args:

~/spark-2.3.0-bin-hadoop2.7/python/pyspark/sql/utils.py in deco(*a, **kw)
     61     def deco(*a, **kw):
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:
     65             s = e.java_exception.toString()

~/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    318                 raise Py4JJavaError(
    319                     "An error occurred while calling 012.\n".
--> 320                     format(target_id, ".", name), value)
    321             else:
    322                 raise Py4JError(

Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 78.0 failed 1 times, most recent failure: Lost task 0.0 in stage 78.0 (TID 78, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/home/lisa/spark-2.3.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/sql/types.py", line 1556, in _getattr_
    idx = self.__fields__.index(item)
ValueError: 'toDF' is not in list

这是什么意思？什么清单？

【问题讨论】：

类似：***.com/questions/40667879/…

如果我们可以看到add_final_score 功能，并且如果您解释您的目标是什么，那么我们可以更好地帮助您

【参考方案1】：

"toDF" 在 DataFrame 上工作，如您在此处看到的： https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=todf#pyspark.sql.DataFrame.toDF

在你的代码中，我猜“exploded”是一个df，所以在你使用“.rdd”之后，它就变成了一个rdd。然后，当您使用“map”时，您会再次返回一个 rdd。

您不能在 rdd 行上应用“toDF”。如果您想将 rdd 转回 DataFrame，您需要使用类似（取决于您的 spark 版本）：https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=createdataframe#pyspark.sql.SparkSession.createDataFrame

此外，您不能使用“map”在 DataFrame 上应用函数，因为 rdd 不能保存它有行的 DataFrame