将RDD列提取到新的RDD中

我有一个包含多个列的rdd，并且想要提取名为“age”的列。我试过了：将RDD列提取到新的RDD中

rddAge = df.rdd.map(lambda row: Row(age=row.age))

代码没有返回错误。但是当我只是试图做一个rddAge.count（），我得到了一个很长的回溯：

Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe. 
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 29.0 failed 10 times, most recent failure: Lost task 0.9 in stage 29.0 (TID 365, yp-spark-dal09-env5-0048): org.apache.spark.api.python.PythonException: Traceback (most recent call last): ...

我做错了吗？

感谢

来源

2017-02-10 Jay

我试过你的代码，并做得很好... 你可以发布你做什么与DF？ –

您不会转储堆栈跟踪的重要组件。 – Chinny84

您是否正在导入行？导入行对我来说工作正常

from pyspark.sql import Row 

rddAge = df.rdd.map(lambda row: Row(age=row.rddAge)) 
rddAge.count()

来源

2017-02-10 06:19:21

在Spark 2，您可以直接select在数据框的顶部栏，在scala相应的代码片段是：

val ageDF = df.select("age") 
ageDF.printSchema()

输出：

root 
|-- age: long (nullable = true)

希望这有助于...

来源

2017-02-10 06:11:44

将RDD列提取到新的RDD中

回答

相关问题