Kafka streams: adding dynamic fields at runtime to avro record

user_1558254 0
Viewed 2385

I want to implement a configurable Kafka stream which reads a row of data and applies a list of transforms. Like applying functions to the fields of the record, renaming fields etc. The stream should be completely configurable so I can specify which transforms should be applied to which field. I'm using Avro to encode the Data as GenericRecords. My problem is that I also need transforms which create new columns. Instead of overwriting the previous value of the field they should append a new field to the record. This means the schema of the record changes. The solution I came up with so far is iterating over the list of transforms first to figure out which fields I need to add to the schema. I then create a new schema with the old fields and new fields combined

The list of transforms(There is always a source field which gets passed to the transform method and the result is then written back to the targetField):

val transforms: List[Transform] = List(
    FieldTransform(field = "referrer", targetField = "referrer", method = "mask"),
    FieldTransform(field = "name", targetField = "name_clean", method = "replaceUmlauts")
)

case class FieldTransform(field: String, targetField: String, method: String)

method to create the new schema, based on the old schema and the list of transforms

def getExtendedSchema(schema: Schema, transforms: List[Transform]): Schema = {    

var newSchema = SchemaBuilder
      .builder(schema.getNamespace)
      .record(schema.getName)
      .fields()

    // create new schema with existing fields from schemas and new fields which are created through transforms
    val fields = schema.getFields ++ getNewFields(schema, transforms)

    fields
      .foldLeft(newSchema)((newSchema, field: Schema.Field) => {
        newSchema
          .name(field.name)
          .`type`(field.schema())
          .noDefault()
          // TODO: find way to differentiate between explicitly set null defaults and fields which have no default
          //.withDefault(field.defaultValue())
      })

     newSchema.endRecord()
   }



 def getNewFields(schema: Schema, transforms: List[Transform]): List[Schema.Field] = {
    transforms
      .filter { // only select targetFields which are not in schema
        case FieldTransform(field, targetField, method) =>  schema.getField(targetField) == null
        case _ => false
      }
      .distinct
    .map { // create new Field object for each targetField
      case FieldTransform(field, targetField, method) =>
      val sourceField = schema.getField(field)
      new Schema.Field(targetField, sourceField.schema(), sourceField.doc(), sourceField.defaultValue())
    }
}

Instantiating a new GenericRecord based on an old record

 val extendedSchema = getExtendedSchema(row.getSchema, transforms)
 val extendedRow = new GenericData.Record(extendedSchema)

 for (field <- row.getSchema.getFields) {
     extendedRow.put(field.name, row.get(field.name))
 }

I tried to look for other solutions but couldn't find any example which had changing data types. It feels to me like there must be a simpler cleaner solution to handle changing Avro schemas at runtime. Any ideas are appreciated.

Thanks, Paul

apache-kafka-streams apache-kafka avro scala
1 Answers

I have implemented Passing Dynamic values to your avro schema and validating union to in schema

Example :-

RestTemplate template = new RestTemplate();
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity<String> entity = new HttpEntity<String>(headers);
        ResponseEntity<String> response = template.exchange(""+registryUrl+"/subjects/"+topic+"/versions/"+version+"", HttpMethod.GET, entity, String.class);
        String responseData = response.getBody();
        JSONObject jsonObject = new JSONObject(responseData); // add your json string which you will pass from postman
        JSONObject jsonObjectResult = new JSONObject(jsonResult);
        String getData = jsonObject.get("schema").toString();
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(getData);
        GenericRecord genericRecord = new GenericData.Record(schema);
        schema.getFields().stream().forEach(field->{
            genericRecord.put(field.name(),jsonObjectResult.get(field.name()));
        });
        GenericDatumReader<GenericRecord>reader = new GenericDatumReader<GenericRecord>(schema);
        boolean data = reader.getData().validate(schema,genericRecord );
user_82150840
Related