非數字條件的dropDuplicates

我有一個看起來像這樣的資料框（我還有幾列，但它們不相關）：

 ----------- ----------- ---------------                                         
|item_id    |location_id|decision       |
 ----------- ----------- --------------- 
|     111111|  A        |        True   |
|     111111|  A        |        False  |
|     111111|  A        |        False  |
|     222222|  B        |        False  |
|     222222|  B        |        False  |
|     333333|  C        |        True   |
|     333333|  C        |        True   |
|     333333|  C        |        Unsure |
 ----------- ----------- --------------- 

我想這樣做dropDuplicates("item_id", "location_id")我可以洗掉具有相同item_idandlocation_id的行，但我想保留包含TrueOR （Unsure如果存在）的行。如果沒有重復行包含Trueor Unsure，False則任何行都可以。對于上面的示例，我希望生成的資料框如下所示：

 ----------- ----------- ---------------                                         
|item_id    |location_id|decision       |
 ----------- ----------- --------------- 
|     111111|  A        |        True   |
|     222222|  B        |        False  |
|     333333|  C        |        Unsure |
 ----------- ----------- --------------- 

對于item_id111111 和location_idA，我想要decisionTrue 的行，因為存在這樣的一行。對于item_id222222 和location_idB，由于沒有一行包含 True，因此選擇其中任何一個都可以。對于item_id333333 和location_idC，所有行都包含所需的值True或Unsure，因此選擇三個中的任何一個都可以。

我正在使用 Scala，因此將不勝感激 Scala 中的解決方案。

uj5u.com熱心網友回復：

這是代碼：

輸入準備：

//spark : My SparkSession
import spark.implicits._
  val df = Seq(
    (111111, "A", "True"),
    (111111, "A", "False"),
    (111111, "A", "False"),
    (222222, "B", "False"),
    (222222, "B", "False"),
    (333333, "C", "True"),
    (333333, "C", "True"),
    (333333, "C", "Unsure")
  ).toDF("item_id", "location_id", "decision")

  df.printSchema()
  /** root
    * |-- item_id: integer (nullable = false)
    * |-- location_id: string (nullable = true)
    * |-- decision: string (nullable = true)
    */

實作所需輸出的代碼：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

//Step 1) create a WindowSpec : MyWindow is same as that of groupBy("item_id", "location_id") 
//but I want to keep track of the order of True, False and Unsure in that partition 
//so, will order my partition based on the col("decision") which is why we have window functions.
  val MyWindow = Window
    .partitionBy(col("item_id"), col("location_id"))
    .orderBy(desc("decision"))

  df
//Step 2) add row_number to each record in that window (based on the mentioned ordering in MyWindow),
//in this case based on the descending order of col("decision")
    .withColumn("row_number", row_number().over(MyWindow))
//Step 3) It turns out we only need first row from each partition  
//based on the decision to select Unsure (then) True (then) False (based on the order of preference),
//so, we filter in only first row.
    .filter(col("row_number").equalTo(1))
    .drop(col("row_number"))
    .orderBy(col("item_id"))
    .show(false)


/**
OUTPUT:
 ------- ----------- -------- 
|item_id|location_id|decision|
 ------- ----------- -------- 
|111111 |A          |True    |
|222222 |B          |False   |
|333333 |C          |Unsure  |
 ------- ----------- -------- 
*/

EDIT1（根據評論）：

改進的代碼（在 WindowSpec 中沒有排序 col("decision")）：

For achieving this, you need to write Custom UserDefinedAggregateFunction for you to have more control over decision attribute range of values, in your requirement it can be like this:

 object MyBestDecisionUDF extends UserDefinedAggregateFunction {

    // step 1) : to set priority score to your decisions which you can configure somewhere
    val decisionOrderMap =
      Map("Unsure" -> 4, "True" -> 3, "False" -> 2, "Zinc" -> 1, "Copper" -> 0)

    /** all overridden functions come from UserDefinedAggregateFunction Abstract Class
      */
    override def inputSchema: StructType = StructType(
      StructField("input_str", StringType, false) :: Nil
    )

    override def bufferSchema: StructType = StructType(
      StructField("buffer_str", StringType, false) :: Nil
    )

    override def dataType: DataType = StringType

    override def deterministic: Boolean = true

    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      buffer.update(0, "")
    }

    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      // main step : updating buffer always to hold best decision string value
      if (
        decisionOrderMap.getOrElse(
          buffer.get(0).toString(),
          -1
        ) < decisionOrderMap.getOrElse(input(0).toString(), -1)
      ) {
        buffer.update(0, input(0))
      }

    }

    override def merge(
        buffer1: MutableAggregationBuffer,
        buffer2: Row
    ): Unit = {}

    override def evaluate(buffer: Row): Any = {
      buffer(0)
    }

  }

  /** ############################################################### 
    * Calling Custom UDAF
    * ###############################################################
    */

/**
INPUT:

 ------- ----------- -------- 
|item_id|location_id|decision|
 ------- ----------- -------- 
|111111 |A          |True    |
|111111 |A          |False   |
|111111 |A          |False   |
|222222 |B          |False   |
|222222 |B          |False   |
|333333 |C          |True    |
|333333 |C          |Unsure  |
|444444 |D          |Copper  |
|444444 |D          |Zinc    |
 ------- ----------- -------- 

*/

  df
  // Custom UDF evaluated column
    .withColumn(
      "my_best_decision",
      MyBestDecisionUDF(col("decision")).over(
        Window
          .partitionBy(col("item_id"), col("location_id"))
      )
    )
    .drop(col("decision"))
    .distinct()
    .orderBy(col("item_id"))
    .show(false)
/**
 * OUTPUT:
 ------- ----------- ---------------- 
|item_id|location_id|my_best_decision|
 ------- ----------- ---------------- 
|111111 |A          |True            |
|222222 |B          |False           |
|333333 |C          |Unsure          |
|444444 |D          |Zinc            |
 ------- ----------- ---------------- 
*/

標籤：斯卡拉 阿帕奇火花 apache-spark-sql

上一篇：如何使用DatabricksNotebook從Key:ValuePair中提取值

下一篇：SparkApplicationMaster是否始終運行在EMR集群的主節點上