在 scala Spark 中连接 UDF

Question

我已经编写了以下代码，运行良好。但是我想连接UDF，这样代码可以压缩成几行。请建议我怎么做。下面是我编写的代码。

val myUdf1 = udf((Number: Long) => ((Number) >> 24) & 255)
val myUdf2 = udf((Number: Long) => ((Number) >> 16) & 255)
val myUdf3 = udf((Number: Long) => ((Number) >> 8) & 255)
val myUdf4 = udf((Number: Long) => (Number) & 255)

val df=Data.withColumn("bitwise 1", myUdf1(Data("Ip")))
  .withColumn("bitwise 2", myUdf2(Data("Ip")))
  .withColumn("bitwise 3", myUdf3(Data("Ip")))
  .withColumn("bitwise 4", myUdf4(Data("Ip")))

val FinalDF =  df.withColumn("FinalIp",concat(col("bitwise 1"),lit("."),col("bitwise 2"),lit("."),col("bitwise 3"),lit("."),col("bitwise 4")))
.drop("bitwise 1").drop("bitwise 2").drop("bitwise 3").drop("bitwise 4")

Answer 1

我认为，这在没有udf的情况下是可以实现的-

val Data = spark.range(2).withColumn("Ip", lit(10))
    val myUdf1 = udf((Number: Long) => ((Number) >> 24) & 255)
    val myUdf2 = udf((Number: Long) => ((Number) >> 16) & 255)
    val myUdf3 = udf((Number: Long) => ((Number) >> 8) & 255)
    val myUdf4 = udf((Number: Long) => (Number) & 255)

    val df=Data.withColumn("bitwise 1", myUdf1(Data("Ip")))
      .withColumn("bitwise 2", myUdf2(Data("Ip")))
      .withColumn("bitwise 3", myUdf3(Data("Ip")))
      .withColumn("bitwise 4", myUdf4(Data("Ip")))

    val FinalDF =  df.withColumn("FinalIp",concat(col("bitwise 1"),lit("."),col("bitwise 2"),lit("."),col("bitwise 3"),lit("."),col("bitwise 4")))
      .drop("bitwise 1").drop("bitwise 2").drop("bitwise 3").drop("bitwise 4")
    FinalDF.show(false)

    /**
      * +---+---+--------+
      * |id |Ip |FinalIp |
      * +---+---+--------+
      * |0  |10 |0.0.0.10|
      * |1  |10 |0.0.0.10|
      * +---+---+--------+
      */

 spark.range(2).withColumn("Ip", lit(10))
      .withColumn("FinalIp",
        concat_ws(".", expr("shiftRight(Ip, 24) & 255"), expr("shiftRight(Ip, 16) & 255"),
          expr("shiftRight(Ip, 8) & 255"), expr("Ip & 255"))
      ).show(false)

    /**
      * +---+---+--------+
      * |id |Ip |FinalIp |
      * +---+---+--------+
      * |0  |10 |0.0.0.10|
      * |1  |10 |0.0.0.10|
      * +---+---+--------+
      */

Answer 2

正如@Someshwar Kale所建议的，你可以没有UDF。

如果选择使用 UDF，则可以将 UDF 中的函数抽象出来，并串联到一个函数中

scala> Data.show
+---+
| ip|
+---+
| 10|
| 20|
+---+


scala> val a:Seq[(Int, Int)] = Seq((24, 255), (16,255), (8, 255),(0,255))
a: Seq[(Int, Int)] = List((24,255), (16,255), (8,255), (0,255))

scala> val myUdf = udf((number: Long) => (a.map((t:(Int, Int)) => (number >> t._1) & t._2).mkString(".")))
myUdf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(LongType)))

scala> Data.withColumn("finalIp", myUdf($"ip")).show
+---+--------+
| ip| finalIp|
+---+--------+
| 10|0.0.0.10|
| 20|0.0.0.20|
+---+--------+