How can I do a join on each array element in a column and replace with something from the join table?

I'd still love to see an inline solution/row level solution, but this is the only real solution I've found so far.

from pyspark.sql.functions import last, posexplode_outer

df_with_fruitids = (

    # Step 1
    base_df
    .select(
        col("Id"),
        col("FruitNames"),
        posexplode_outer(col("FruitNames")).alias("Position", "Fruit")
    ).alias("exploded")
    
    # Step 2
    .join(lookup_df.alias("lookup"), col("lookup.Name") == col("exploded.Fruit"), how="left")

    # Step 3
    .withColumn(
        "FruitIds",
        when(
            col("FruitNames").isNull(), lit(None)
        )
        .otherwise(
            collect_list(col("lookup.Id"))
            .over(
                Window
                .partitionBy(col("exploded.Id"))
                .orderBy(col("Position"))
            )
        )
    )

    # Step 4
    .groupBy(col("exploded.Id"))
    .agg(last(col("FruitIds")).alias("FruitIds"))

    # Step 5
    .join(base_df, "Id", how="right")
)

Step 1

Break out base_df into only the necessary columns Id and FruitNames along with posexplode_outer to explode with an ordinal (the "pos" part means positional and the "outer" part means it includes null/empty list). This generates the following dataframe I have aliased as exploded:

Step 2

Join this dataframe with lookup_df on lookup.name == exploded.Fruit.

Step 3

Then, add a column that is null if FruitNames is null, otherwise, it is collect_list with a window function over exploded.Id (the Id from base_df) and ordered on the ordinal position Position from the posexplode_outer. This generates the following dataframe:

Step 4

Since only the last row of the window function contains all the values, on the group by, use last function in the aggregation to grab the final version per each id.

Step 5

After, just join it back on base_df on Id using a right join to get the rest of the columns.

from pyspark.sql import functions as func

base_df.alias("df1").join(
    lookup_df.alias("df2"), 
    on=func.array_contains(func.col("df1.FruitNames"), func.col("df2.Name")),
    how="left"
).groupBy(
    "df1.Id", "df1.FruitNames"
).agg(
    func.map_from_arrays(func.collect_list("df2.Name"), func.collect_list("df2.Id")).alias("Reference")
).select(
    "Id",
    "FruitNames",
    "Reference",
    func.expr("transform(FruitNames, x -> Reference[x])").alias("values")
).show(
    10, False
)

You can use the conditional joining to check if the reference table key exist in the FruitNames array or not, then create a map column as a reference to get the value.

+---+-----------------------+--------------------------------------+---------+
|Id |FruitNames             |Reference                             |values   |
+---+-----------------------+--------------------------------------+---------+
|1  |[apple, banana, orange]|{apple -> 1, banana -> 2, orange -> 3}|[1, 2, 3]|
|2  |[banana]               |{banana -> 2}                         |[2]      |
|3  |[grape, apple]         |{apple -> 1, grape -> 4}              |[4, 1]   |
|4  |[]                     |{}                                    |[]       |
|5  |null                   |{}                                    |null     |
|6  |[orange, grape]        |{orange -> 3, grape -> 4}             |[3, 4]   |
+---+-----------------------+--------------------------------------+---------+

---

Edit: 2025-07-07

In my previous answer, I didn't handle remaining columns (i.e. Col1, Col2, ..., Col99). I use a struct column to store and unnest it later.

from pyspark.sql import functions as func

base_df.select(
    "Id", "FruitNames", func.struct([col for col in base_df.columns if "Col" in col]).alias("ColStruct")
).alias("df1").join(
    lookup_df.alias("df2"), 
    on=func.array_contains(func.col("df1.FruitNames"), func.col("df2.Name")),
    how="left"
).groupBy(
    "df1.Id", "df1.FruitNames", "df1.ColStruct"
).agg(
    func.map_from_arrays(func.collect_list("df2.Name"), func.collect_list("df2.Id")).alias("Reference")
).select(
    "Id",
    "FruitNames",
    "Reference",
    func.expr("transform(FruitNames, x -> Reference[x])").alias("Values"),
    "ColStruct.*"
).orderBy(
    "Id"
).show(
    10, False
)