import io.archivesunleashed._
import io.archivesunleashed.df._

val target = udf((vs: Any) => {
   				       var res = ""
   					if(vs != null){
   						res = vs.toString.split(",")(1)
   					}
   					res
   				})
val src = udf((vs: Any) => {
   				var res = ""
   				if(vs != null){
   					val s = vs.toString.split(",")(0)
   					if(s.length() != 0)
   						res = s.drop(1)
   				}
   				res
   			   })
val modify = udf((str: String) => str.replaceAll("^\\\\s*www\\\\.", ""))

val df = RecordLoader.loadArchives("./src/test/resources/arc/example.arc.gz",sc).webpages()
   				 .select($"crawl_date",explode_outer(ExtractLinksDF($"url",$"content")).as("link"))

df.select($"crawl_date",modify(ExtractDomainDF(src($"link"))).as("Source"),modify(ExtractDomainDF(target($"link"))).as("Target"))
.filter($"Source" =!= "")
.filter($"Target" =!= "")
.groupBy("crawl_date","Source","Target")
.count()
.filter($"count" > 5)
.orderBy(desc("count"))
.show(20)