apache-spark - 使用列值作为 spark DataFrame 函数的参数
问题描述
考虑以下数据框:
#+------+---+
#|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 函数来复制这种行为?
解决方案
一种选择是使用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|
#+------+---+-----+
推荐阅读
- django - 如何使用 {% include %} 或任何其他方法在 django 的主页上添加功能联系表?
- .net-core - 在 Azure Pipeline 中转换应用设置
- python - python selenium webdriver chrome
- java - 如何将 picoCLI 项目变成命令?
- ruby - 无法加载此类文件 -- plist Catalina 和 Xcode 12.4
- java - 如何使用 Glide 从 JSON 中获取图像
- reactjs - 使用 JSX 调用子组件会抛出错误
- c# - 找不到类型或命名空间名称“AutoCompleteEntities2”(您是否缺少 using 指令或程序集引用?)
- emacs-prelude - 在 Mx prelude-update 之后出现错误:错误的类型参数:stringp,100
- python - 在python中按深度间隔绘制数据