Как извлечь ключ / значение из карт в отдельные столбцы в Spark SQL

Question

У меня есть таблица с картой. Я хочу сделать 2 отдельных столбца из этой карты - 1. столбец ключей 2. столбец значений.

input.show();

+---------------------+--------------------+--------------+----------------+---------------+--------------+-----------------------+
|addedSkuWithTimestamp|     fbaSKUAdditions|fbaSKURemovals|      merchantId|mfnSKUAdditions|mfnSKURemovals|removedSkuWithTimestamp|
+---------------------+--------------------+--------------+----------------+---------------+--------------+-----------------------+
| [Test1 -> 1234567...|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|   A4QA5OYD4Y45F|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|ANOTHER_MERCHANT|             []|          null|                   null|
|                 null|[CLAM_SUNGLASS_CA...|            []|ANOTHER_MERCHANT|             []|          null|                   null|
+---------------------+--------------------+--------------+----------------+---------------+--------------+-----------------------+

Но я хочу выводить как

test1  123456789 

Test2  123456780 

Как получить 2 разных столбца (ключевой столбец и столбец значений) с карты?

Dataset<Row> removed_skus = input
                    .withColumn("sku", functions.explode(input.col("removedSkuWithTimestamp")))
                    .withColumn("skuType", functions.lit("MFN"))
                    .select(input.col("merchantId").alias("merchant_id"), new Column("sku").,
                            new Column("skuType"))
                    .distinct()
                    .groupBy("merchant_id")
                    .agg(functions.collect_list("sku").alias("removedSkus"));

Answer 1

Сначала давайте создадим некоторые данные:

val df = Seq(
    (Map("sku1"->"timestamp1"), "AFN"),
    (Map("sku2"->"timestamp2"), "AFN"),
    (null, "AFN") 
).toDF("addedSkuWithTimestamp", "skuType")
.show(false)

+---------------------+-------+
|addedSkuWithTimestamp|skuType|
+---------------------+-------+
| [sku1 -> timestamp1]|    AFN|
| [sku2 -> timestamp2]|    AFN|
|                 null|    AFN|
+---------------------+-------+

Это будет иметь следующую схему:

scala> df.printSchema()

root
 |-- addedSkuWithTimestamp: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- skuType: string (nullable = true)

Искра <2,3 </strong>

Следующий код извлечет столбцы sku_key и sku_value из AddedSkuWithTimestamp столбца, используя функцию mapToTupleUDF udf. :

val mapToTupleUDF = udf((sku: Map[String, String]) => if(sku != null) sku.toSeq(0) else null)

df.withColumn("addedSkuWithTimestamp", mapToTupleUDF($"addedSkuWithTimestamp"))
  .withColumn("Sku", when($"addedSkuWithTimestamp".isNotNull, $"addedSkuWithTimestamp._1"))
  .withColumn("Timestamp", when($"addedSkuWithTimestamp".isNotNull, $"addedSkuWithTimestamp._2"))
  .show(false)

+---------------------+-------+----+----------+
|addedSkuWithTimestamp|skuType|Sku |Timestamp |
+---------------------+-------+----+----------+
|[sku1, timestamp1]   |AFN    |sku1|timestamp1|
|[sku2, timestamp2]   |AFN    |sku2|timestamp2|
|null                 |AFN    |null|null      |
+---------------------+-------+----+----------+

Обратите внимание, что мы можем получить доступ к addedSkuWithTimestamp._1, только если addedSkuWithTimestamp не равен нулю .

Искра> = 2,3

Начиная с Spark 2.3.0 вы можете использовать встроенные map_values и map_keys:

df.withColumn("Sku", map_keys($"addedSkuWithTimestamp").getItem(0))
  .withColumn("Timestamp", map_values($"addedSkuWithTimestamp").getItem(0))
  .show(false)

Выход:

+---------------------+-------+----+----------+
|addedSkuWithTimestamp|skuType|Sku |Timestamp |
+---------------------+-------+----+----------+
|[sku1 -> timestamp1] |AFN    |sku1|timestamp1|
|[sku2 -> timestamp2] |AFN    |sku2|timestamp2|
|null                 |AFN    |null|null      |
+---------------------+-------+----+----------+