Scala UDF转PySpark运行抛出TypeError:int对象不可迭代异常

你的PySpark转换代码共有3处问题，直接触发运行异常：

• 遍历逻辑写反：Scala原逻辑是遍历tokensToSearchFor（对应传入的name字段）的每个字符做存在性判断，你的代码写成遍历tokensToSearchIn（对应传入的age字段），逻辑完全颠倒。
• 入参类型未做兼容：原Scala代码中addVectors方法声明两个入参均为String类型，传入Int类型的age字段时，Spark会通过隐式转换自动将Int转为String；但PySpark不会执行该自动转换，传入的age值是整数类型，你在列表推导式中尝试遍历整数，就会抛出TypeError: 'int' object is not iterable错误。
• UDF返回类型声明错误：原Scala方法返回Seq[Int]即整数序列，你注册UDF时声明返回类型为ArrayType(StringType())（字符串数组），和实际返回的整数列表类型不匹配，即使修复前两个问题，这里也会触发类型转换异常。

先导入依赖的类型类，再修复上述三个问题即可，运行逻辑和结果与Scala原代码完全一致：

from pyspark.sql.functions import udf, col
from pyspark.sql.types import ArrayType, IntegerType

def getVectors(searchTermsToProcessWithTokens): 

    def addVectors(tokensToSearchFor: str, tokensToSearchIn):
        # 强制将第二个入参转为字符串，对齐Scala的隐式转换行为
        search_in_str = str(tokensToSearchIn)
        # 遍历第一个入参的每个字符，和原Scala逻辑保持一致
        return [1 if char in search_in_str else 0 for char in tokensToSearchFor]
      
    # 修正UDF返回类型为整数数组，匹配原Seq[Int]定义
    addVectorsUdf = udf(addVectors, ArrayType(IntegerType()))

    res_df = searchTermsToProcessWithTokens \
      .withColumn("search_term_vector", addVectorsUdf(col("name"), col("age"))) \
      .withColumn("keyword_text_vector", addVectorsUdf(col("name"), col("age")))
      
    return res_df

注：修正后代码的运行结果和Scala原代码完全对齐：生成的向量长度等于name字段的字符串长度，由于name一般为字母组成、age转成字符串后为数字，二者无重叠字符，因此返回全0向量，和你描述的Scala运行结果一致。

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