使用 Spark 数据帧 API 计算列中的特定字符

Question

我在 Spark 数据帧df中有一列包含位。列的格式为字符串：

10001010000000100000000000000000
10001010000000100000000100000000

有没有一种简单有效的方法来创建新列"no_of_ones"使用数据帧计算列的频率？使用RDD，我可以map(lambda x:x.count('1'))（PySpark）.
此外，如何检索包含列表位置的列表？

Answer 1

我能想到的一种方法是删除所有零，然后计算字段的长度。

df.show
+--------------------+
|          bytestring|
+--------------------+
|10001010000000100...|
|10001010000000100...|
+--------------------+

df.withColumn("no_of_ones" , length(regexp_replace($"bytestring", "0", "")) ).show
+--------------------+----------+
|          bytestring|no_of_ones|
+--------------------+----------+
|10001010000000100...|         4|
|10001010000000100...|         5|
+--------------------+----------+

Answer 2

一般来说，当你在 （py）spark SQL 的预定义函数中找不到你需要的东西时，你可以编写一个用户定义函数 （UDF），它可以做任何你想做的事情（参见 UDF）。

请注意，在您的情况下，编码良好的 udf 可能比 scala 或 java 中的正则表达式解决方案更快，因为您不需要实例化新字符串并编译正则表达式（for 循环就可以了）。但是，在pyspark中可能会慢得多，因为在执行器上执行python代码总是会严重损害性能。

Answer 3

由于它是二进制（0/1），因此上述答案将起作用。感谢奥利的回答。

但以防万一您需要在具有 .

例如：

Find '~' in a string "ABCDE~FGH~IJAK~123~$$$" 

选择以下解决方案。

df.withColumn("no_of_ones" , length($"bytestring") - length(regexp_replace($"bytestring", "~", "")) ).show

Answer 4

Spark 3.4+的出现次数regexp_count。

F.expr(r"regexp_count(col_name, '1')")

^{注意：特殊字符需要使用\进行转义，例如 F.expr(r"regexp_count(col_name, '\+')")}

计数的完整示例：

from pyspark.sql import functions as F
df = spark.createDataFrame(
    [("10001010000000100000000000000000",),
     ("10001010000000100000000100000000",)],
    ["bytestring"])
df.withColumn('no_of_ones', F.expr(r"regexp_count(bytestring, '1')")).show()
# +--------------------+----------+
# |          bytestring|no_of_ones|
# +--------------------+----------+
# |10001010000000100...|         4|
# |10001010000000100...|         5|
# +--------------------+----------+

<小时 />

对于这些职位，我可以建议一种更复杂的方法，涉及高阶函数transform和filter。

seq = "sequence(1, length(bytestring))"
located = F.expr(f"transform({seq}, x -> locate('1', bytestring, x))")
cleaned = F.filter(F.array_distinct(located), lambda x: x != 0)
df.withColumn('pos_of_ones', cleaned).show(truncate=0)
# +--------------------------------+-----------------+
# |bytestring                      |pos_of_ones      |
# +--------------------------------+-----------------+
# |10001010000000100000000000000000|[1, 5, 7, 15]    |
# |10001010000000100000000100000000|[1, 5, 7, 15, 24]|
# +--------------------------------+-----------------+