Spark Scala下如何为DataFrame列实现Luhn校验并修复代码报错

错误原因分析 #

你遇到的两个报错本质是混淆了Spark Column类型和Scala原生数据类型的用法：

• 第一个报错：org.apache.spark.sql.Column是Spark SQL的逻辑列表达式，代表整个列的计算逻辑，不是行级的具体值，没有reverse这类Scala字符串/集合的原生方法。你当前写的逻辑是试图在Driver端定义方法时就直接计算sum，完全没有按行处理数据。
• 第二个报错：===是Spark Column类型专用的等值比较运算符，你的sum是Scala原生Int类型，直接用===会触发类型不匹配错误，原生数值比较应该用==，但这个错误是第一个错误的衍生问题。

解决方案 #

使用Spark UDF（用户自定义函数）封装Luhn校验逻辑，UDF会自动将每行的列值作为入参传入校验逻辑，操作的是原生Scala字符串类型，符合你的预期。

完整实现代码 #

import org.apache.spark.sql.Column
import org.apache.spark.sql.functions.udf

// 原生Scala实现Luhn校验逻辑，入参是行级的实际字符串值
def luhnCheck(s: String): Boolean = {
  // 非法输入直接返回不通过
  if (s == null || s.isEmpty || !s.matches("\\d+")) return false
  var odd = true
  var sum = 0
  for (c <- s.reverse) {
    val digit = c.asDigit
    if (odd) {
      sum += digit
    } else {
      val double = digit * 2
      sum += (double % 10) + (double / 10)
    }
    odd = !odd
  }
  sum % 10 == 0
}

// 注册为Spark UDF
val luhnCheckUDF = udf(luhnCheck _)

使用示例 #

假设你的DataFrame名为sourceDF，待校验的字符串列名为num_str，新增校验列的代码如下：

val resultDF = sourceDF.withColumn("luhn_pass", luhnCheckUDF(col("num_str")))

新增的luhn_pass列为Boolean类型，校验通过为true，不通过为false，符合你的需求。

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