Использование значения столбца в качестве параметра искровой функции DataFrame - PullRequest
Новая
       82

Использование значения столбца в качестве параметра искровой функции DataFrame

0 голосов
/ 02 июля 2018

Рассмотрим следующий фрейм данных: 

#+------+---+
#|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?

1 Ответ

0 голосов
/ 10 сентября 2018

Один из вариантов - использовать pyspark.sql.functions.expr, что позволяет использовать значения столбцов в качестве входных данных для функций spark-sql.

На основании комментария @ user8371915 Я обнаружил, что следующие работы: 

from pyspark.sql.functions import expr

df.select(
    '*',
    expr('posexplode(split(repeat(",", rpt), ","))').alias("index", "col")
).where('index > 0').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|
#+------+---+-----+
...