PySpark DataFrame连续字符出现次数统计的报错解决方法

错误原因分析 #

你遇到的TypeError: Column is not iterable是因为直接将PySpark的Column对象传入了普通Python函数。PySpark的Column是分布式数据的抽象，不是本地可迭代的Python字符串，普通Python函数无法直接操作它。必须通过**用户定义函数（UDF）**将Python逻辑包装成PySpark能处理的列操作，或者使用PySpark内置函数实现逻辑。

解决方案一：修正并使用UDF #

步骤1：修正Python函数逻辑 #

你的本地函数逻辑是对的，但迁移到PySpark时，函数需要接收本地字符串而非Column对象，同时修正循环索引的错误（原代码中for i in name是遍历字符，而非索引，会导致逻辑错误）：

def code_func(name):
    if not name:
        return ""
    count = 1
    strng = ""
    for i in range(len(name)-1):
        if name[i] == name[i+1]:
            count += 1
        else:
            strng += name[i] + str(count)
            count = 1
        # 处理最后一个字符
        if i == len(name)-2:
            if name[i] != name[i+1]:
                strng += name[i+1] + "1"
            else:
                strng += name[i] + str(count)
    return strng

步骤2：注册为PySpark UDF #

将上述函数注册为UDF，指定返回类型为StringType，然后应用到DataFrame列：

完整代码：

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

if __name__ == "__main__":
    my_conf = SparkConf()
    my_conf.set("spark.app.name","my 1st app")
    my_conf.set("spark.master","local[*]")

    spark = SparkSession.builder.config(conf=my_conf).getOrCreate()

    def code_func(name):
        if not name:
            return ""
        count = 1
        strng = ""
        for i in range(len(name)-1):
            if name[i] == name[i+1]:
                count += 1
            else:
                strng += name[i] + str(count)
                count = 1
            if i == len(name)-2:
                if name[i] != name[i+1]:
                    strng += name[i+1] + "1"
                else:
                    strng += name[i] + str(count)
        return strng

    # 注册UDF
    code_udf = udf(code_func, StringType())

    # 读取数据
    df = spark.read.format("csv").option("path", "C:/Users/hp/OneDrive/Desktop/ddd.txt").load().toDF("words")
    # 应用UDF生成新列
    df2 = df.withColumn("coded_words", code_udf(df["words"]))
    df2.show()

解决方案二：使用PySpark内置函数（无UDF，性能更优） #

对于分布式数据，使用PySpark内置函数比Python UDF性能更好，因为内置函数是JVM执行的，避免了Python-JVM的序列化开销。可以通过拆分字符串、窗口函数分组计数、拼接结果实现：

from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    split, explode, count, concat_ws, concat,
    col, lit, lag, sum, row_number
)
from pyspark.sql.window import Window

if __name__ == "__main__":
    my_conf = SparkConf()
    my_conf.set("spark.app.name","my 1st app")
    my_conf.set("spark.master","local[*]")

    spark = SparkSession.builder.config(conf=my_conf).getOrCreate()

    # 读取数据
    df = spark.read.format("csv").option("path", "C:/Users/hp/OneDrive/Desktop/ddd.txt").load().toDF("words")

    # 步骤1：将字符串拆分为单个字符的数组，并展开为行，记录字符位置
    df_exploded = df.select(
        col("words"),
        explode(split(col("words"), "(?!^)")).alias("char"),
        row_number().over(Window.partitionBy("words").orderBy(lit(1))).alias("pos")
    )

    # 步骤2：标记连续相同字符的分组ID
    window_spec = Window.partitionBy("words").orderBy("pos")
    df_grouped = df_exploded.withColumn(
        "group_id",
        sum(
            (col("char") != lag(col("char"), 1).over(window_spec)).cast("int")
        ).over(window_spec)
    ).fillna(0, subset=["group_id"])

    # 步骤3：按分组统计字符出现次数
    df_agg = df_grouped.groupBy("words", "group_id", "char").agg(
        count("*").alias("count")
    ).orderBy("words", "group_id")

    # 步骤4：拼接每个分组的字符与计数，生成最终结果
    df_result = df_agg.groupBy("words").agg(
        concat_ws("", concat(col("char"), col("count"))).alias("coded_words")
    )

    df_result.show()

输出结果 #

两种方法都会得到你期望的输出：

+-----------+-----------+
|      words|coded_words|
+-----------+-----------+
|   aaabbcca|   a3b2c2a1|
|  aabbbccaa|   a2b3c2a2|
|       abcd|   a1b1c1d1|
|    dddeert|   d3e2r1t1|
|aaabbbacccd| a3b3a1c3d1|
+-----------+-----------+

内容的提问来源于stack exchange，提问作者Nitish