您需要充分了解的是，df上的所有标准、非聚合操作一次只能看到一行。所以，你必须从一行或一个集合的Angular 来考虑。在单行上下文中对整个df调用函数 withColumn 这个电话很不寻常。当您要将行与其他行进行比较/合并时，通常会将表与另一个表或表本身联接，然后使用合并的表。
你想要的是类似的东西

// create DF with similarity scores AND customer, which is treated as row id here
val similarityDF = flattenData.map(row => row.getAs[String]("customer") -> countSimilarColumns(row, referenceCustomerRow)).
                   toDF("customer","similarity_score")
// join your original DF with similarityDF based on "customer"
flattenData.join(simlarityDF, usingColumn = "customer").show()

这允许您匹配两个df的行。
udf在这里并不真正适用，因为与内置函数类似，udf应用于一个（或多个）特定列以生成值（可以使用 withColumn )，您要查看整排。

flattenData 属于类型 DataFrame &在数据上应用map函数可以得到 Dataset .
你通过了考试的成绩 flattenData.map(row => countSimilarColumns(row, referenceCustomerRow)) 至 withColumn 但是 withColumn 只能获取类型的数据 org.apache.spark.sql.Column 所以，如果你想添加以上结果 UDF 必须使用的列 collect 函数并将其传递给 lit 请检查下面的代码。

flattenData
.withColumn("similarity_score",lit(
        flattenData
        .map(row => countSimilarColumns(row, referenceCustomerRow))
        .collect
        .map(_.toInt)
   )
)

基于样本数据，添加以下逻辑。

scala> df.show(false)
+-----+-----+-----+-----+----------+
|att-a|att-b|att-c|att-d|customer  |
+-----+-----+-----+-----+----------+
|7    |3    |10   |10   |customer-1|
|9    |7    |12   |4    |customer-2|
|7    |3    |1    |10   |customer-3|
|9    |14   |10   |4    |customer-4|
+-----+-----+-----+-----+----------+
scala> val conditions = df.columns.filterNot(_ == "customer").map(c => (when(count(col(c)).over(Window.partitionBy(col(c)).orderBy(col(c).asc)) =!= 1,lit(1)).otherwise(0))).reduce(_ + _) // if row_number is 1 then adding 1 else 0 ..
conditions: org.apache.spark.sql.Column = (((CASE WHEN (NOT (count(att-a) OVER (PARTITION BY att-a ORDER BY att-a ASC NULLS FIRST unspecifiedframe$()) = 1)) THEN 1 ELSE 0 END + CASE WHEN (NOT (count(att-b) OVER (PARTITION BY att-b ORDER BY att-b ASC NULLS FIRST unspecifiedframe$()) = 1)) THEN 1 ELSE 0 END) + CASE WHEN (NOT (count(att-c) OVER (PARTITION BY att-c ORDER BY att-c ASC NULLS FIRST unspecifiedframe$()) = 1)) THEN 1 ELSE 0 END) + CASE WHEN (NOT (count(att-d) OVER (PARTITION BY att-d ORDER BY att-d ASC NULLS FIRST unspecifiedframe$()) = 1)) THEN 1 ELSE 0 END)

最终结果

scala> df.withColumn("similarity_score",conditions).show(false)
+-----+-----+-----+-----+----------+----------------+
|att-a|att-b|att-c|att-d|customer  |similarity_score|
+-----+-----+-----+-----+----------+----------------+
|9    |7    |12   |4    |customer-2|2               |
|7    |3    |1    |10   |customer-3|3               |
|7    |3    |10   |10   |customer-1|4               |
|9    |14   |10   |4    |customer-4|3               |
+-----+-----+-----+-----+----------+----------------+