I have a Spark Streaming Job which processes messages coming from Kafka.
My incoming json that I process sort of looks like
{"sv" : 1.0, "field1" : "some data"}
The only thing I do is put these in a MYSQL database.
However, I need to process these messages differently based on the schema version number!
For instance, I may get data that looks like the below in the same stream
{"sv" : 1.0, "field1" : "some data"}
{"sv" : 1.1, "field1" : "some data", "field2" : "new data"}
{"sv" : 1.2, "field1" : "some data", "field2" : "new data", "field3" : "data"}
Now what I do is I have a function that formats the data for me like so
def formatData(json: String): Option[Data] = {
var outputData: Option[Data] = None
val jsonObject = new JSONObject(json)
outputData = formatDataBasedOnSchemaVersion(jsonObject)
outputData
}
and another function that formats based on a schema version number
private def formatDataBasedOnSchemaVersion(jsonObject: JSONObject): Option[Data] = {
val outputData = {
jsonObject.getDouble("sv") match {
case 1.0 => Some(formatVersion_1_0(jsonObject))
case 1.1 => Some(formatVersion_1_1(jsonObject))
case 1.2 => Some(formatVersion_1_2(jsonObject))
case x: Double => logger.warn("No formatter found for schema version: " + x); None
}
}
outputData
}
An example of my format function can look like
private def formatVersion_1_2(jsonObject: JSONObject): Data = {
val f1 = jsonObject.getString("field1")
val f2 = jsonObject.getString("field2")
val f3 = jsonObject.getString("field3")
val data = Data(f1,f2,f3)
data
}
In the format_1_0 function, all I do is pull out the "field1" parameter.
My Data class is simple DTO it just looks like
case class Data(field1: String, field2: String, field3: String)
If I get schema version 1.0, field2 and field3 are left blank and inserted into the DB as blank values.
The problem is, I have to hard code in the schema version numbers like "1.0", "1.1" etc.. and design a new method to pull out the extra fields. So for every schema change, I have to edit the code and add a new method to pull out the new data. So is there any better pattern I can use that can handle this? Or maybe a framework? I've heard of ORM would this help with that problem or would I still need to make similar code changes for schema version changes?