PySpark - 从列表中获取字符串位置

Posted 2023-04-15

【中文标题】PySpark - 从列表中获取字符串位置

【英文标题】：PySpark - get string position from list

【发布时间】：2020-09-08 01:39:49

【问题描述】：

我有一个包含 FN 列的数据框和这些列值子集的列表 例如

    **FN**
    ABC
    DEF
    GHI
    JKL
    MNO
    
    List:
    ["GHI","DEF"]

我想在我的数据框中添加一列，如果列值存在于列表中，我记录列表中的位置，即我的结束 DF

    FN     POS
    ABC
    DEF    1
    GHI    0
    JKL
    MNO

我的代码如下

    from pyspark.sql.functions import udf
    from pyspark.sql.types import StringType
    l = ["GHI","DEF"]

    x = udf(lambda fn, p = l: p.index(fn), StringType())
    
    df = df.withColumn('POS', when(col("FN").isin(l), x(col("FN"))).otherwise(lit('')))

但是在运行时，我得到一个“Job aborted due to stage failure”异常和一系列其他异常，唯一有意义的部分是“ValueError: 'JKL' is not in list”（JKL 是我的 DF 列中的随机其他列）

如果我只输入“fn”而不是“p.index(fn)”，我会在新列中获得正确的列值，同样，如果我使用“p.index("DEF")”，我会返回“1”，因此这些单独工作，有什么想法为什么会出现异常？

TIA

---

编辑：我已经设法通过在 lambda 中执行 if-else 来解决这个问题，这几乎意味着它在 withColumn 语句中的“isin”检查之前执行 lambda。 我想知道什么（除了上述是否属实），有没有人对如何以更好的方式实现这一点有更好的建议？

【问题讨论】：

请发布预期输出

您将不得不使用 if-else，因为如果项目不在列表中，您需要处理

创建地图map1 = expr(f"""map(','.join(f"'e',i" for i,e in enumerate(List)))""")，然后执行df.withColumn('POS', coalesce(map1[col('FN')],lit(''))).show()。确保从 pyspark.sql.functions 导入 expr、coalesce、lit

【参考方案1】：

这是我的尝试。我已经为给定列表制作了一个数据框并加入它们。

from pyspark.sql.functions import *

l = ['GHI','DEF']
m = [(l[i], i) for i in range(0, len(l))]

df2 = spark.createDataFrame(m).toDF('FN', 'POS')
df1 = spark.createDataFrame(['POS','ABC','DEF','GHI','JKL','MNO'], "string").toDF('FN')

df1.join(df2, ['FN'], 'left').show()

+---+----+
| FN| POS|
+---+----+
|JKL|null|
|MNO|null|
|DEF|   1|
|POS|null|
|GHI|   0|
|ABC|null|
+---+----+