Spark Job Pattern

ActivityInsightsJob.scala

class ActivityInsightsJob(activityReader: SparkReader,
                          analyticsReader: SparkReader,
                          insightsWriter: SparkWriter
                         )(implicit val sparkSession: SparkSession) extends SparkTask {

    def run(): Unit = {
        val metricsDF = analyticsReader.read(Some(AnalyticsSchema))
          .transform(Events.isActivityImpression)
          .transform(Events.isActivityView)
          .transform(Events.isBookmarked)
          .transform(withCountersBy("activityId"))
          .transform(withRatingScore)

        val activityInsightsDF = activityReader.read(Some(ActivitySchema))
          .transform(cleanActivity)
          .transform(enrichActivityWithMetrics(metricsDF))
          .transform(withHashId("activityId"))
          .transform(withNow)

        insightsWriter.write(activityInsightsDF, mode = Append)
  }
}