#+------+---+
#|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| #+------+---+-----+
这可以产生正确的答案。但是，我无法使用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 楼
一种选择是使用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| #+------+---+-----+评论
有趣……一如既往！
–thebluephantom
19年11月19日在21:48
评论
f.expr（“”“ repeat（”，“，rpt）”“”）代替f.repeat（“，”，f.col（“ rpt”）））？@ user8371915 df.select（'*'，f.expr（'posexplode（split（repeat（“，”，rpt），“，”））'）。alias（“ index”，“ col”））。where（ 'index> 0'）。drop（“ col”）。sort（'letter'，'index'）。show（）有效。您知道这是否是使用Column作为参数的唯一方法吗？为什么它可以使用sql语法？
@ user8371915，请考虑发布您的建议作为答案（可以从我的问题中对其进行编辑）。我认为将来对其他人有好处。