如何在 PySpark 数据框中添加具有替代值的列？

Question

我有以下示例数据框

df = spark.createDataFrame([('start','end'), ('start1','end1')] ,["start", "end"])

我想展开每一行中的值，并在生成的行中关联交替的 1-0 值。这样我就可以识别每行中的 start/end 个条目。

我可以通过这种方式达到预期的效果

from pyspark.sql.window import Window
w = Window().orderBy(lit('A'))
df = (df.withColumn('start_end', fn.array('start', 'end'))
        .withColumn('date', fn.explode('start_end'))
        .withColumn('row_num', fn.row_number().over(w)))
df = (df.withColumn('is_start', fn.when(fn.col('row_num')%2 == 0, 0).otherwise(1))
        .select('date', 'is_start'))

这给出了

| date   | is_start |
|--------|----------|
| start  | 1        |
| end    | 0        |
| start1 | 1        |
| end1   | 0        |

但对于这样一个简单的任务来说似乎过于复杂了。

有没有不使用UDF的better/cleaner方法？

Answer 1

你可以试试union:

df = spark.createDataFrame([('start','end'), ('start1','end1')] ,["start", "end"])
df = df.withColumn('startv', F.lit(1))
df = df.withColumn('endv', F.lit(0))
df = df.select(['start', 'startv']).union(df.select(['end', 'endv']))
df.show()

+------+------+
| start|startv|
+------+------+
| start|     1|
|start1|     1|
|   end|     0|
|  end1|     0|
+------+------+

您可以重命名列并从此处开始对行重新排序。

Answer 2

我在用例中遇到过类似情况。在我的情况下，我有巨大的数据集（~50GB）并且进行任何自我 join/heavy 转换会导致更多的内存和不稳定的执行。

我将数据集降低了一层，并使用了 rdd 的平面图。这将使用映射端转换，并且在随机播放、cpu 和内存方面具有成本效益。

df = spark.createDataFrame([('start','end'), ('start1','end1')] ,["start", "end"])
df.show()
+------+----+
| start| end|
+------+----+
| start| end|
|start1|end1|
+------+----+

final_df = df.rdd.flatMap(lambda row: [(row.start, 1), (row.end, 0)]).toDF(['date', 'is_start'])
final_df.show()
+------+--------+
|  date|is_start|
+------+--------+
| start|       1|
|   end|       0|
|start1|       1|
|  end1|       0|
+------+--------+

Answer 3

您可以使用 pyspark.sql.functions.posexplode along with pyspark.sql.functions.array.

首先从 start 和 end 列中创建一个数组，然后用位置分解它：

from pyspark.sql.functions import array, posexplode

df.select(posexplode(array("end", "start")).alias("is_start", "date")).show()
#+--------+------+
#|is_start|  date|
#+--------+------+
#|       0|   end|
#|       1| start|
#|       0|  end1|
#|       1|start1|
#+--------+------+

如何在 PySpark 数据框中添加具有替代值的列？

How to add column with alternate values in PySpark dataframe?

apache-spark

pyspark

pyspark-sql