CODEWITHSUNDEEP
apache-sparkpyspark
arxidomourhs
arxidomourhs

Reputation: 19

Mapping key and list of values to key value using pyspark

I have a dataset which consists of two columns C1 and C2.The columns are associated with a relation of many to many.

What I would like to do is find for each C2 the value C1 which has the most associations with C2 values overall.

For example:

C1  | C2
 1  | 2
 1  | 5
 1  | 9
 2  | 9
 2  | 8

We can see here that 1 is matched to 3 values of C2 while 2 is matched to 2 so i would like as output:

   Out1 |Out2| matches
     2  | 1  | 3
     5  | 1  | 3
     9  | 1  | 3 (1 wins because 3>2)
     8  | 2  | 2

What I have done so far is:

  dataset = sc.textFile("...").\
          map(lambda line: (line.split(",")[0],list(line.split(",")[1]) ) ).\
          reduceByKey(lambda x , y : x+y )

What this does is for each C1 value gather all the C2 matches,the count of this list is our desired matches column. What I would like now is somehow use each value in this list as a new key and have a mapping like :

(Key ,Value_list[value1,value2,...]) -->(value1 , key ),(value2 , key)...

How could this be done using spark? Any advice would be really helpful.

Thanks in advance!

Upvotes: 1

Views: 440

Answers (2)

Mohana B C
Mohana B C

Reputation: 5487

We can get the desired result easily using dataframe API.

from pyspark.sql import *
import pyspark.sql.functions as fun
from pyspark.sql.window import Window

spark = SparkSession.builder.master("local[*]").getOrCreate()

# preparing sample dataframe
data = [(1, 2), (1, 5), (1, 9), (2, 9), (2, 8)]
schema = ["c1", "c2"]
df = spark.createDataFrame(data, schema)
 
output = df.withColumn("matches", fun.count("c1").over(Window.partitionBy("c1"))) \
.groupby(fun.col('C2').alias('out1')) \
.agg(fun.first(fun.col("c1")).alias("out2"), fun.max("matches").alias("matches"))

output.show()

# output
+----+----+-------+
|Out1|out2|matches|
+----+----+-------+
|   9|   1|      3|
|   5|   1|      3|
|   8|   2|      2|
|   2|   1|      3|
+----+----+-------+

Upvotes: 1

mck
mck

Reputation: 42332

The dataframe API is perhaps easier for this kind of task. You can group by C1, get the count, then group by C2, and get the value of C1 that corresponds to the highest number of matches.

import pyspark.sql.functions as F

df = spark.read.csv('file.csv', header=True, inferSchema=True)

df2 = (df.groupBy('C1')
         .count()
         .join(df, 'C1')
         .groupBy(F.col('C2').alias('Out1'))
         .agg(
           F.max(
             F.struct(F.col('count').alias('matches'), F.col('C1').alias('Out2'))
           ).alias('c')
         )
         .select('Out1', 'c.Out2', 'c.matches')
         .orderBy('Out1')
      )

df2.show()
+----+----+-------+
|Out1|Out2|matches|
+----+----+-------+
|   2|   1|      3|
|   5|   1|      3|
|   8|   2|      2|
|   9|   1|      3|
+----+----+-------+

Upvotes: 1

Related Questions