Pivot Kafka KTable results using flatMap

Viewed 483

I've got two JSON documents as follows:

[
  {
    "ProductID": 12,
    "ProductName": "Product 1",
    "CountryID": 55,
    "CountryName": "Country 1",
    "Companies": [{
        "CompanyID": 1,
        "CompanyName": "Company 1"
      }, {
        "CompanyID": 2,
        "CompanyName": "Company 2"
      }
    ]
  },
  {
    "ProductID": 13,
    "ProductName": "Product 2",
    "CountryID": 55,
    "CountryName": "Country 1",
    "Companies": [{
        "CompanyID": 1,
        "CompanyName": "Company 1"
      }, {
        "CompanyID": 2,
        "CompanyName": "Company 2"
      }
    ]
  }
]

And this is supposed to be a KTable, not a KStream because some records will be deleted.

Key is (ProductID, CountryID).

I'd like to pivot this data and have (CompanyID) as a key together with every (ProductID, CountryID) combinations in an array in such manner:

[
  {
    "CompanyID": 1,
    "CompanyName": "Company 1",
    "ProductsCountries": [{
        "ProductID": 12,
        "CountryID": 55
      }, {
        "ProductID": 13,
        "CountryID": 55
      }
    ]
  },
  {
    "CompanyID": 2,
    "CompanyName": "Company 2",
    "ProductsCountries": [{
        "ProductID": 12,
        "CountryID": 55
      }, {
        "ProductID": 13,
        "CountryID": 55
      }
    ]
  }
]

And whenever a (ProductID, CountryID) combination disappears from my KTable, I'd like to remove it from array where (CompanyID) has it.

This seems to be semi-possible using KStream because I could just flatMap it and add new (ProductID, CountryID) combinations to my document, however I won't be able to capture deletes.

Is there a way to do it with KTable?

0 Answers
Related