expire.scala

// Returns Some((user_id, email_content)) if an email should be 
// triggered None otherwise
def eventTriggeredEmail(userId: Int, eventDescription: String): 
  Option[(Int, String)] = ???

object Emailer extends App with MyConfigurationMixin {
    val sparkConf = new SparkConf()
      .setAppName("Emailer")
      .setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(1))

    // Tuple is user_id, email
    val userRecords: JdbcRDD[(Int, String)] = GetUsersFromDatabase()

    // Tuple is user_id, event json
    val events: DStream[(Int, String)] = ConnectToExternalEventStream()

    val emails: DStream[(Int, String)] = events
      .flatMap(eventTriggeredEmail.tupled)

    emails
      .transform(rdd => rdd.join(userRecords))      
      .foreach { case (userId, (eventBody, emailAddress)) => 
        sendEmail(emailAddress, emailBody)
      }
}