Simplest way to go about transforming data from kafka

Question

I am working on a project that pulls data from multiple db sources using kafka connect. I want to then be able to transform the data into a specified json format and then finally push that final json to an S3 bucket preferably using kafka connect to keep my overhead down.

Here is an example of what the data currently looks like coming into kafka (in avro format):

{"tableName":"TABLE1","SchemaName{"string":"dbo"},"tableID":1639117030,"columnName":{"string":"DATASET"},"ordinalPosition":{"int":1},"isNullable":{"int":1},"dataType":{"string":"varchar"},"maxLength":{"int":510},"precision":{"int":0},"scale":{"int":0},"isPrimaryKey":{"int":0},"tableSizeKB":{"long":72}}
{"tableName":"dtproperties","SchemaName":{"string":"dbo"},"tableID":1745441292,"columnName":{"string":"id"},"ordinalPosition":{"int":1},"isNullable":{"int":0},"dataType":{"string":"int"},"maxLength":{"int":4},"precision":{"int":10},"scale":{"int":0},"isPrimaryKey":{"int":1},"tableSizeKB":{"long":24}}

This looks like so when converted to JSON:

{
      "tablename" : "AS_LOOKUPS",
      "tableID": 5835333,
      "columnName": "SVALUE",
      "ordinalPosition": 6,
      "isNullable": 1,
      "dataType": "varchar",
      "maxLength": 4000,
      "precision": 0,
      "scale": 0,
      "isPrimaryKey": 0,
      "tableSize": 0,
      "sizeUnit": "GB"
},
{
      "tablename" : "AS_LOOKUPS",
      "tableID": 5835333,
      "columnName": "SORT_ORDER",
      "ordinalPosition": 7,
      "isNullable": 1,
      "dataType": "int",
      "maxLength": 4,
      "precision": 10,
      "scale": 0,
      "isPrimaryKey": 0,
      "tableSize": 0,
      "sizeUnit": "GB"
}

My goal is to get the data to look like so:

{
  "header": "Database Inventory",
  "DBName": "DB",
  "ServerName": "server@server.com",
  "SchemaName": "DBE",
  "DB Owner": "Name",
  "DB Guardian" : "Name/Group",
  "ASV" : "ASVC1AUTODWH",
  "ENVCI": "ENVC1AUTODWHORE",
  "Service Owner" : "Name/Group",
  "Business Owner" : "Name/Group",
  "Support Owner" : "Name/Group",
  "Date of Data" : "2017-06-28 12:12:55.000",
  "TABLE_METADATA": {
  "TABLE_SIZE" : "500",
  "UNIT_SIZE" : "GB",
  "TABLE_ID": 117575457,
  "TABLE_NAME": "spt_fallback_db",
  "COLUMN_METADATA": [
  {
    "COLUMN_NM": "xserver_name",
    "DATE_TYPE": "varchar",
    "MAX_LENGTH": 30,
    "PRECISION": 0,
    "SCALE": 0,
    "IS_NULLABLE": 0,
    "PRIMARY_KEY": 0,
    "ORDINAL_POSITION": 1
  },
  {
    "COLUMN_NM": "xdttm_ins",
    "DATE_TYPE": "datetime",
    "MAX_LENGTH": 8,
    "PRECISION": 23,
    "SCALE": 3,
    "IS_NULLABLE": 0,
    "PRIMARY_KEY": 0,
    "ORDINAL_POSITION": 2
  }, ........

The header data will mostly be generic, but some of the stuff like date and etc. will need to be populated.

Initially my original thought were that I could do everything utilizing kafka connect, and that I could just create a schema for the way I want the data to be formatted. I am having a problem though with utilizing a different schema with the connectors and I'm not really sure if it is even possible.

Another solution I thought about was utilizing Kafka Streams, and writing code to transform the data into what is needed. I'm not to sure how easy it is do that w/ Kafka Streaming.

And finally a third solution I have seen is to utilize Apache Spark, and manipulating the data with dataframes. But this will add more overhead.

I'm honestly not to sure what route to go, or if any of these solutions are what I'm looking for. So I am open to all advice on how to solve this problem.

Simplest way to go about transforming data from kafka

1 Answers