Рассмотрим следующий фрейм данных:
#+------+---+
#|letter|rpt|
#+------+---+
#| X| 3|
#| Y| 1|
#| Z| 2|
#+------+---+
, который можно создать с помощью следующего кода:
df = spark.createDataFrame([("X", 3),("Y", 1),("Z", 2)], ["letter", "rpt"])
Предположим, я хотел повторить в каждой строке количество раз, указанное в столбце rpt
, как в этом вопросе .
Один из способов - скопировать мое решение на этот вопрос, используя следующий запрос pyspark-sql
:
query = """
SELECT *
FROM
(SELECT DISTINCT *,
posexplode(split(repeat(",", rpt), ",")) AS (index, col)
FROM df) AS a
WHERE index > 0
"""
query = query.replace("\n", " ") # replace newlines with spaces, avoid EOF error
spark.sql(query).drop("col").sort('letter', 'index').show()
#+------+---+-----+
#|letter|rpt|index|
#+------+---+-----+
#| X| 3| 1|
#| X| 3| 2|
#| X| 3| 3|
#| Y| 1| 1|
#| Z| 2| 1|
#| Z| 2| 2|
#+------+---+-----+
Это работает и дает правильный ответ. Однако я не могу воспроизвести это поведение с помощью функций API DataFrame.
Я пытался:
import pyspark.sql.functions as f
df.select(
f.posexplode(f.split(f.repeat(",", f.col("rpt")), ",")).alias("index", "col")
).show()
Но это приводит к:
TypeError: 'Column' object is not callable
Почему я могу передать столбец в качестве ввода в repeat
в запросе, но не из API? Есть ли способ воспроизвести это поведение, используя функции DataFrame spark?