使用列值作为火花DataFrame函数的参数
使用列值作为火花DataFrame函数的参数
提问于 2018-07-02 16:41:28
考虑以下DataFrame：
#+------+---+
#|letter|rpt|
#+------+---+
#| X| 3|
#| Y| 1|
#| Z| 2|
#+------+---+它可以使用以下代码创建：
df = spark.createDataFrame([("X", 3),("Y", 1),("Z", 2)], ["letter", "rpt"])假设我想重复每一行在列rpt中指定的次数，就像在这个question中一样。
一种方法是使用以下solution查询将我的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|
#+------+---+-----+这是有效的，并产生正确的答案。但是，我无法使用DataFrame API函数复制此行为。
我试过：
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中传递？是否有一种方法可以使用spark DataFrame函数复制此行为？
回答 1
Stack Overflow用户
回答已采纳
发布于 2018-09-10 20:04:24
一种选择是使用pyspark.sql.functions.expr，它允许您使用列值作为输入来激发-sql函数。
基于@ Based 8371915的comment，我发现了以下工作：
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|
#+------+---+-----+页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/51140470
复制相关文章
相似问题
腾讯云开发者
