change.mongodb.scala

def process(changeDocument: ChangeStreamDocument[Document]) = {
  implicit val formats: DefaultFormats.type = DefaultFormats
  val rdd = spark.sparkContext.makeRDD(List(write(mapRecord(changeDocument))))

  if (rddQueue.isEmpty)
    rddQueue.enqueue(rdd)
  else
    rddQueue.enqueue(rddQueue.dequeue().union(rdd))
}

case class MongoChangeRecord(token: String, document: String, operationType: String, removedFields: Seq[String], updatedFields: String, timestamp: Long)
def mapRecord(doc: ChangeStreamDocument[Document]): MongoChangeRecord = {
  MongoChangeRecord(
    doc.getResumeToken.toJson,
    doc.getFullDocument.toJson,
    doc.getOperationType.getValue,
    Try(doc.getUpdateDescription.getRemovedFields.asScala).getOrElse(List.empty),
    Try(doc.getUpdateDescription.getUpdatedFields.toJson).getOrElse(""),
    Calendar.getInstance().getTimeInMillis
  )
}