如何在spark scala中做df列的Luhn检查

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

def luhnCheck(col: Column): Column = {
  val digits = col.cast(StringType).rlike("^[0-9]+$")
  val reversed = reverse(col.cast(StringType))
  val doubled = when(row_number().over(Window.orderBy(lit(1))) % 2 === 0, reversed.cast(IntegerType) * 2)
    .otherwise(reversed.cast(IntegerType))
  val summed = sum(doubled % 10 + doubled / 10)
  val valid = (summed % 10 === 0)
  when(digits && valid, lit(true)).otherwise(lit(false))
}

val df = spark.read.format("csv").option("header", "true").load("path/to/file.csv")
val checkedDF = df.withColumn("luhn_check", luhnCheck(col("column_name")))