I am new to spark. I have two dataframes df1 and df2. df1 has three rows. df2 has more than few million rows. I want to check whether all items in df2 are in transaction of df1, if so sum up the costs. Example as follows:

 df1 = spark.createDataFrame(
 [(1, [(1), (4), (2) ,(3)])],
 ("id", "transaction")

 df2 = spark.createDataFrame(
 [([ (1),(2),(3)], 2.0), ([(5),(2),(3)], 1.0) ],
 ("items", "cost")

Desired Result is:

 id   transaction score
  1    [1,4,2,3]    2.0

My Current code is:


def check(trans,itm):
   itertrans = iter(trans)
   return all(i in itertrans for i in itm)


The other solution which i have:

 costs = (df1
# Explode transaction
.select("id", explode("transactions").alias("item"))
        # Add id so we can later use it to identify source
        .withColumn("_id", monotonically_increasing_id().alias("_id"))
         # Explode items
            "_id", explode("items").alias("item"), 
            # We'll need size of the original items later
            size("items").alias("size"), "cost"), 
 # Count matches in groups id, items
 .groupBy("_id", "id", "size", "cost")
 # Compute cost
 .agg(sum_(when(col("size") == col("count"), col("cost"))).alias("score")))

with 500,000 rows of df2, crossJoin method produced results quicker (in 6 min) when compared to the explode and join soution (which took 15 mins).

with 900,000 rows of df2, both solution run for hours without results. I run the program in standalone mode.


How can i improve the speed? Any other solution? if i convert to rdd and run in clutser, will the process speeds up? How to convert to rdd?(i tried but don't know how to convert the group by and aggreagta part)

Any help would be great.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.