import io.archivesunleashed._
import io.archivesunleashed.app._
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\\\\.", ""))

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


df = 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"))

val Columns = Seq("crawl_date","Source","Target","count")
df = df.toDF(Columns:_*)

WriteGraph.nodesWithIdsDF(df).show(20)

WriteGraph.edgeNodesDF(df).show(10)

WriteGraph.asGraphmlDF(df,"output.csv")