This can already be done with the existing kafkaimporter if the names of the tables are known before they need to be created. The module creates the tables using CREATE TABLE IF NOT EXISTS to allow the users to modify the created table if it needs to be. The minimum requirements for the table is that there's a field named data in it into which JSON documents can be inserted.

The limitations on the data can be removed by either creating the table before the KafkaImporter is started, in which case the CREATE TABLE IF NOT EXISTS does nothing, or by altering the structure of the existing table. The minimum requirement that must be met is that the table contains the data field to which string values can be inserted into.

Added the engine option which makes the storage engine configurable.