Spark 2.4.x: Duplicate keys in map

user_6765102 0
Viewed 664

Given the following two arrays columns in Spark:

df = spark.createDataFrame(
    [
        Row(
            [Row(x=0, y='a'), Row(x=1, y='b'), Row(x=0, y='c')]
        )
    ],
    ["col"]
)
df.show()

+------------------------+
|col                     |
+------------------------+
|[[0, a], [1, b], [0, c]]|
+------------------------+

I want to create a map using map_from_entries:

df.withColumn(
    "mapped", f.map_from_entries("col")
).show(truncate=False)


+------------------------+------------------------+
|col                     |mapped                  |
+------------------------+------------------------+
|[[0, a], [1, b], [0, c]]|[0 -> a, 1 -> b, 0 -> c]|
+------------------------+------------------------+

In Spark 2.4.x, the behavior of the map functions, is to duplicate the keys. In the example, we have two values (a, c) for the same key 0.

I am aware that Spark 3.x behaves differently using the configuration: spark.sql.mapKeyDedupPolicy=LAST_WIN, the results of the same operation will be:

+------------------------+----------------+
|col                     |mapped          |
+------------------------+----------------+
|[{0, a}, {1, b}, {0, c}]|{0 -> c, 1 -> b}|
+------------------------+----------------+

where only the second value c for the key 0 is kept in the map.

I am wondering if there is a way to obtain the same results using Spark 2.4.

pyspark apache-spark-sql apache-spark python
2 Answers

Another way you can try is to convert to RDD then back to dataframe.

df = df.withColumn("mapped", F.map_from_entries("col")).rdd.toDF()
df.show(truncate=False)

+------------------------+----------------+
|col                     |mapped          |
+------------------------+----------------+
|[[0, a], [1, b], [0, c]]|[0 -> c, 1 -> b]|
+------------------------+----------------+
user_94778430

A possible solutions as suggested by Steven is to use an udf:

s = t.MapType(t.IntegerType(), t.StringType())
@f.udf(s)
def fix_map(s):
    return s

df.withColumn(
    "mapped", fix_map(f.map_from_entries("col"))
).show(truncate=False)


+------------------------+----------------+
|col                     |mapped          |
+------------------------+----------------+
|[[0, a], [1, b], [0, c]]|[0 -> c, 1 -> b]|
+------------------------+----------------+

The map passed to the UDF in Python is converted to a a Python dictionary, which doesn't allow to have duplicated keys. The drawback is the overhead required to exchange data between the JVM and the Python interpreter.

user_67651020
Related