python - 使用 pyspark 将 RDD 行转换为数据帧时出错
问题描述
我编写了一个要应用于数据帧的函数,但首先我必须将数据帧转换为要映射的 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 {0}{1}{2}.\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
这是什么意思?什么清单?
解决方案
“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
推荐阅读
- regex - 选择不同名称的正则表达式
- mysql - 从 EAV 中选择返回行两次
- laravel-5 - Laravel 类在生产中找不到,但适用于本地环境
- c - 将for循环中的值存储到数组中?
- docker - 如何创建 docker 内部注册表
- robotframework - 如何根据机器人框架中关键字的结果实现If条件?
- python - MySQL:如何编写查询以选择 MAX 和今天的日期以从给定列表中获取数据并存储在数据库中
- java - 在 EditText 中动态更改光标颜色
- docker - Docker 烧瓶应用程序请求源标头已删除
- css - 在 Firefox 中使用动画 3d CSS 变换时的伪影