scala – 具有新指标列的Spark Dataframe Group

发布时间：2020-12-16 18:13:31 所属栏目：安全来源：网络整理

导读：我需要按“KEY”列进行分组,需要检查“TYPE_CODE”列是否同时包含“PL”和“JL”值,如果是,那么我需要添加一个指标列为“Y”,否则为“N” 示例： //Input Values val values = List(List("66","PL"),List("67","JL"),"PO"),List("68","PO")).map(x =(x(0),x(

我需要按“KEY”列进行分组,需要检查“TYPE_CODE”列是否同时包含“PL”和“JL”值,如果是,那么我需要添加一个指标列为“Y”,否则为“N”

示例：

//Input Values
    val values = List(List("66","PL"),List("67","JL"),"PO"),List("68","PO")).map(x =>(x(0),x(1)))

    import spark.implicits._
    //created a dataframe
    val cmc = values.toDF("KEY","TYPE_CODE")

    cmc.show(false)
    ------------------------
    KEY |TYPE_CODE  |
    ------------------------
    66  |PL |
    67  |JL |
    67  |PL |
    67  |PO |
    68  |JL |
    68  |PO |
    -------------------------

预期产出：

对于每个“KEY”,如果它具有“TYPE_CODE”则同时具有PL& JL然后Y.
否则N.

-----------------------------------------------------
    KEY |TYPE_CODE  | Indicator
    -----------------------------------------------------
    66  |PL         | N
    67  |JL         | Y
    67  |PL         | Y
    67  |PO         | Y
    68  |JL         | N
    68  |PO         | N
    ---------------------------------------------------

例如,
67有PL& JL – 所以“Y”
66只有PL – 所以“N”
68只有JL – 所以“N”

解决方法

一种选择：

1)收集TYPE_CODE作为列表;

2)检查它是否包含特定字符串;

3)然后用爆炸展平列表：

(cmc.groupBy("KEY")
    .agg(collect_list("TYPE_CODE").as("TYPE_CODE"))
    .withColumn("Indicator",when(array_contains($"TYPE_CODE","PL") && array_contains($"TYPE_CODE","Y").otherwise("N"))
    .withColumn("TYPE_CODE",explode($"TYPE_CODE"))).show
+---+---------+---------+
|KEY|TYPE_CODE|Indicator|
+---+---------+---------+
| 68|       JL|        N|
| 68|       PO|        N|    
| 67|       JL|        Y|
| 67|       PL|        Y|
| 67|       PO|        Y|
| 66|       PL|        N|
+---+---------+---------+

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!