Events

sealed trait Event
case class ClickEvent(blaBla: String) extends Event
case class ViewEvent(blaBla: String) extends Event

val input: RDD[String] = sc.textFile("logs/events/*")
val events: RDD[Event] = input.parse

val views: RDD[ViewEvent] = events.flatMap {
    case v: ViewEvent => Some(v)
    case _ => None
}

val clicks: RDD[ClickEvent] = events.flatMap {
    case c: ClickEvent => Some(c)
    case _ => None
}

views.saveAsTextFile("views")
clicks.saveAsTextFile("clicks")