将 pyspark.rdd.PipelinedRDD 对象转换为 Pyspark 数据框时获取空值

Question

我的数据集有一列名为“eventAction”。

它具有“conversion”、“purchase”、“check-out”等值。我想将此列转换为将转换映射到 1 并将所有其他类别映射到 0。

我是这样使用 lambda 函数的：

e1 = event1.rdd.map(lambda x: 1 if x.eventAction == 'conversion' else 0)

其中 event1 是我的 spark 数据框的名称。

当打印 e1 时，我明白了：

print(e1.take(5))
[0, 0, 0, 0, 0]

所以我认为 lambda 函数工作正常。现在，当我转换为 pyspark 数据框时，我得到如下所示的空值：

schema1 = StructType([StructField('conversion',IntegerType(),True)])
df = spark.createDataFrame(data=[e1],schema=schema1)
df.printSchema()
df.show()

如果你能帮助我，那就太好了。

谢谢！

Answer 1

spark.createDataFrame 需要 Row 的 RDD，而不是整数的 RDD。在转换为数据框之前，您需要将 RDD map 转换为 Row 对象。请注意，无需在e1 周围添加方括号。

from pyspark.sql import Row

e1 = event1.rdd.map(lambda x: 1 if x.eventAction == 'conversion' else 0).map(lambda x: Row(x))
schema1 = StructType([StructField('conversion',IntegerType(),True)])
df = spark.createDataFrame(data=e1,schema=schema1)

---

也就是说，您尝试做的事情应该可以使用 Spark SQL when 函数轻松完成。无需将 RDD 与自定义 lambda 函数一起使用。例如

import pyspark.sql.functions as F

df = events.select(F.when(F.col('eventAction') == 'conversion', 1).otherwise(0).alias('conversion'))