Как добавить коллекцию столбцов на основе максимальных и минимальных значений в кадре данных

Question

У меня есть это DataFrame

val for_df = Seq((5,7,"5k-7k"),(4,8,"4k-8k"),(6,12,"6k-2k")).toDF("min","max","salary")

Я хочу преобразовать 5k-7k в 5,6,7 и 4k-8k в 4,5,6,7,8.

Оригинал DataFrame:

Желаемый DataFrame

a.select("min","max","salary")
      .as[(Integer,Integer,String)]
      .map{
        case(min,max,salary) =>
          (min,max,salary.split("-").flatMap(x => {
            for(i <- 0 to x.length-1) yield (i)
          }))
      }.toDF("1","2","3").show()

Answer 1

Вы можете легко сделать это с помощью udf.

// The following defines a udf in spark which create a list as per your requirement.
val makeRangeLists = udf( (min: Int, max: Int) => List.range(min, max+1) )

val input = sc.parallelize(List((5,7,"5k-7k"),
                          (4,8,"4k-8k"),(6,12,"6k-12k"))).toDF("min","max","salary")
// Create a new column using the UDF and pass the max and min columns.
input.withColumn("salary_level", makeRangeLists($"min", $"max")).show

Answer 2

Здесь один быстрый вариант с UDF

 import org.apache.spark.sql.functions


  val toSalary = functions.udf((value: String) => {
    val array = value.filterNot(_ == 'k').split("-").map(_.trim.toInt).sorted
    val (startSalary, endSalary) = (array.headOption, array.tail.headOption)

    (startSalary, endSalary) match {
      case (Some(s), Some(e)) => (s to e).toList.mkString(",")
      case _ =>  ""
    }
  })

for_df.withColumn("salary_level", toSalary($"salary")).drop("salary")

Ввод

+---+---+------+
|min|max|salary|
+---+---+------+
|  5|  7| 5k-7k|
|  4|  8| 4k-8k|
|  6| 12| 6k-2k|
+---+---+------+

Результат

+---+---+------------+
|min|max|salary_level|
+---+---+------------+
|  5|  7|       5,6,7|
|  4|  8|   4,5,6,7,8|
|  6| 12|   2,3,4,5,6|
+---+---+------------+

Сначала вы удаляете k и разбиваете строку по тире.Затем вы получаете start и endSalary и выполняете диапазон между ними.

Answer 3

вам нужно создать UDF, чтобы расширить пределы.Следующие UDF преобразуют 5k-7k в 5,6,7 и 4k-8k в 4,5,6,7,8 и так далее

import org.apache.spark.sql.functions._
val inputDF = sc.parallelize(List((5,7,"5k-7k"),(4,8,"4k-8k"),(6,12,"6k-12k"))).toDF("min","max","salary")

val extendUDF = udf((str: String) => {
  val nums = str.replace("k","").split("-").map(_.toInt)
  (nums(0) to nums(1)).toList.mkString(",")
})

val output = inputDF.withColumn("salary_level", extendUDF($"salary"))

Выходные данные:

scala> output.show
+---+---+------+----------------+
|min|max|salary|    salary_level|
+---+---+------+----------------+
|  5|  7| 5k-7k|           5,6,7|
|  4|  8| 4k-8k|       4,5,6,7,8|
|  6| 12|6k-12k|6,7,8,9,10,11,12|
+---+---+------+----------------+