Design pattern that can be used for checking and handling a change in version number for incoming JSON message into data stream?

Question

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?

Answer 1

To the best of my knowledge, ORMs don't solve this problem, but some do have a lot of built in mechanisms for handling migrations for you, but likely not in this kind of context.

What you're creating right now is considered a dispatch table, and I have yet to find a more elegant way to handle them than what you have. There are dynamic ways which "work" so long as a method exists matching the type, but these create a confusing code base that is hard for others to pick up and learn.

You expressed the fundamental issue of this problem, "I have to hard code in the schema version and design a new method to pull out the extra fields". Unfortunately, that problem doesn't leave any room for extra abstraction as I see it. It is at it's simplest level. There are many different ways to organize the code that implements that basic statement, but they will all boil down to "I support this set of schema versions, and here is how each one will be handled" which is exactly what your code is already doing. No amount of different methods, classes, or extra syntactic sugar will remove the basic decision structure you have above, it will just move it around.

The only thing you could possibly do differently, based on your statement that each new method only pulls out new fields, would be to incrementally parse the data object. This means that you have a set of handlers for the data object specific to a schema version, and you apply each one to the data object, each pulling out specific fields that were added in that schema version only. This assumes that existing fields will never be removed or changed in new schema versions, which is probably more limiting than you may want, and again this is a fundamental change of the original statement of each schema version needing to be handled individually.