Spark/Scala iterating two dataframes

1
votes

I am new to spark/scala world. I have two sources of data

  1. Traffic data which has URLs and hostnames

  2. Attribute data which defines rules on the traffic urls. The rules are regex patterns to match the domain name. There could be one or many rules for one attribute-id.

If an URL meets criteria I have to assign an attribute-id. Each row in the traffic can match zero or more attribute conditions

A sample input traffic-data

 visitor_id | url                      
 1000-abc10 | www.motor.com/index.html
 2000-fe30a | www.lifestyle.com/cooking/pasta.html

`attribute-data

attribute_id | rule                               | describtion
101          | motor.com, auto*.com, vehicles.com | "vehicles"
102          | motor.com                          | "auto site"

Expected output:

visitor_id  | attribute_id
1000-abc10  | 101
1000-abc10  | 202

I tried the following :

val traffic_df = spark.read.parquet(<traffic-path>).as[Traffic]
val attribute_df = spark.read.parquet(<attribute-path>).as[Attribute]

traffic_df.map(row => attribute_df.map(r => TrafficAttribute(row.visitor_id, r.attribute_id)))
2
scalaapache-sparkapache-spark-sql
You should add examples of your inputs, what output you want and what is not working with your code - Fabich
I have added the input and output as suggested. - Jai Prakash
I could figure out a solution for my problem I have posted. To complete the loop I am here posting the code snippet below which worked for me. - Jai Prakash

2 Answers

2
votes
case class Traffic(visitor_id: String, page_url : String)
case class ConfigRow(attribute_id: String, rule: String, description: String)
case class OutputRow(visitor_id: String, attribute_id)

val configList = spark.sqlContext.read.json(<config-path>).as[ConfigRow].collect().toList
val trafficDF = spark.read.json(<traffic-path>).as[Traffic]

def determineAttributes(row: Traffic, configList: List[ConfigRow]): ListBuffer[String] = {
    val attributeList = new ListBuffer[String]
    for (c <- configList) {
      rule = c.rule;
      if (<rule matches>) attributeList += c.attribute_id
   }
   attributeList
}

for r = trafficDF.flatMap((row:Traffic) => {
   for (attributeId <- determineAttributes(row, configList)) yield {
      OutputRow(row.visitor_id, attributeId)
   }
})
1
votes

You can use a join with a special join condition on 2 datasets :

val joinCondition = $"a.url".contains($"b.rule")
var joinedDf = trafficDf.as('a).join(attributeDf.as('b),joinCondition)
joinedDf.show()

+----------+--------------------+------------+---------+-----------+
|visitor_id|                 url|attribute_id|     rule|describtion|
+----------+--------------------+------------+---------+-----------+
|1000-abc10|www.motor.com/ind...|         101|motor.com|   vehicles|
|1000-abc10|www.motor.com/ind...|         102|motor.com|  auto site|
+----------+--------------------+------------+---------+-----------+

then you can select the desired column with joinedDf.select("visitor_id","attribute_id")