// Gets top 100 pageviews for the month of May, for Indic-centric language wikipedias and English wikipedia
val card: Int = 100;

def sortBySecond(p1: (String, Long), p2: (String, Long)): Boolean = {
  p1._2 - p2._2 > 0
}

def topMap(top: Vector[(String, Long)], elem: (String, Long)): Vector[(String, Long)] = {
  (top :+ elem).sortWith(sortBySecond).take(card)
}
def topRed(t1: Vector[(String, Long)], t2: Vector[(String, Long)]): Vector[(String, Long)] = {
  (t1 ++ t2).sortWith(sortBySecond).take(card)
}

def tuplify(r: org.apache.spark.sql.Row): (String, (String, Long)) = {
    (r.getString(0), (r.getString(1), r.getLong(2)))
}


val df = sqlContext.parquetFile("/user/joal/pageviews/hourly2")
val sample = df.filter("year = 2015 and month = 5")
val filtered = sample.filter("agent_type = 'user' and project in ('as.wikipedia', 'bn.wikipedia', 'bh.wikipedia', 'bpy.wikipedia', 'en.wikipedia', 'gu.wikipedia', 'hi.wikipedia', 'kn.wikipedia', 'ks.wikipedia', 'ml.wikipedia', 'mr.wikipedia', 'ne.wikipedia', 'new.wikipedia', 'or.wikipedia', 'pi.wikipedia', 'pa.wikipedia', 'sa.wikipedia', 'sd.wikipedia', 'ta.wikipedia', 'te.wikipedia', 'ur.wikipedia')")
val counts = filtered.groupBy($"project", $"page_title").agg($"project", $"page_title", sum($"count") as "count")