PySpark 列到其值的 RDD

Question

我正在寻找将数据框列转换为 RDD 的最直接和惯用的方法。假设列views 包含浮点数。以下是不是我要找的

views = df_filtered.select("views").rdd

因为我最终得到了 RDD[Row] 而不是 RDD[Float]，因此我无法将它提供给 mllib.stat 中的任何 stat 方法（如果我正确理解发生了什么）：

corr = Statistics.corr(views, likes, method="pearson")
TypeError: float() argument must be a string or a number

在 pandas 中，我会使用 .values() 将此 pandas Series 转换为其值的数组，但 RDD .values() 方法似乎无法以这种方式工作。我终于找到了以下解决方案

views = df_filtered.select("views").rdd.map(lambda r: r["views"])

但我想知道是否有更直接的解决方案

Answer 1

您需要为此使用 flatMap。

>>> newdf=df.select("emp_salary")
>>> newdf.show();
+----------+
|emp_salary|
+----------+
|     50000|
|     10000|
|    810000|
|      5500|
|      5500|
+----------+

>>> rdd=newdf.rdd.flatMap(lambda x:x)
>>> rdd.take(10);
[50000, 10000, 810000, 5500, 5500]

你看起来像这样吗？

是的，而不是将您的陈述转换为：

views = df_filtered.select("views").rdd.flatMap(lambda x:x)

Answer 2

使用 RDD 'Dataframe' 的下一个更高抽象层，您可以做到这一点。

from pyspark import SparkContext
from pyspark import SQLContext
from pyspark.sql.types import FloatType
import pandas as pd

#data creation (for example)
dictonary = {'views': [1.902, 2.34334, 0.3434], 'some_other_column':[1,2,3]}
df = pd.DataFrame(data=dictonary)

#create spark context
sc = SparkContext("local", "First App1")
sql = SQLContext(sc)

#create spark dataframe from pandas dataframe
spark_df = sql.createDataFrame(df['views'], FloatType())
spark_rdd = spark_df.rdd

可能有一种不那么繁琐的方法，但这可能会给你一些启发。请记住，RDD 是不可变的。