Migration

Migration.scala

val customerEvents = new JdbcRDD(sc, () => { DriverManager.getConnection(mysqlJdbcString)},
    "select * from customer_events ce, staff, store where ce.store = store.store_name and ce.staff = staff.name " +
        "and ce.id >= ? and ce.id <= ?", startingId, highestId, numberOfPartitions,
    (r: ResultSet) => {
      (r.getString("customer"),
        r.getTimestamp("time"),
        UUID.randomUUID(),
        r.getString("event_type"),
        r.getString("store_name"),
        r.getString("location"),
        r.getString("store_type"),
        r.getString("staff"),
        r.getString("job_title")
        )
    })

  customerEvents.saveToCassandra("test", "customer_events",
      SomeColumns("customer_id", "time", "id", "event_type", "store_name", "store_type", "store_location", "staff_name", "staff_title"))