Connectors → Apache Kafka V2

About Apache Kafka

Apache Kafka is an open-source distributed event streaming platform used for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

About the Apache Kafka V2 connector

With the new version of the Apache Kafka connector, you can create an external data source for one or more Kafka topics. The Kafka topics must contain messages in a valid JavaScript Object Notation (JSON) format.

The new connector provides multiple enhancements over the older connector version, including:

• An enhanced and streamlined user experience
• Supporting multiple topics per data source
• Automatic discovery; no need to generate an Avro schema
• Transformation of JSON messages (Flattening objects and arrays)
• Enhanced performance on Cloud installations and enhanced throughput
• Consuming messages in a batch mode during the load job
• Native support for Data Agent
• Multi-loader support (No need to define the Loader Service for the Kafka Consumer Service)

The Apache Kafka V2 connector supports the following Incorta-specific functionalities:

Steps to connect Apache Kafka and Incorta

To connect Apache Kafka and Incorta, here are the high-level steps, tools, and procedures:

• Create an external data source
• Create a physical schema with the Schema Wizard
• or, Create a physical schema with the Schema Designer
• Load the physical schema
• Explore the physical schema

Create an external data source

Here are the steps to create an external data source with the Apache Kafka V2 connector:

• Sign in to Incorta.
• In the Navigation bar, select Data.
• In the Action bar, select + New → Add Data Source.
• In the Choose a Data Source dialog, in Streaming Data, select Kafka V2.
• In the New Data Source dialog, specify the applicable connector properties.
• To test, select Test Connection.
• Select Ok to save your changes.

Kafka V2 connector properties

Here are the properties of the Kafka V2 connector:

Supported extra options

The following table lists the extra options supported in the Kafka V2 connector.

Create a physical schema with the Schema Wizard

Here are the steps to create a physical schema with the Schema Wizard:

• Sign in to Incorta.
• In the navigation bar, select Schema.
• In the action bar, select + New → Schema Wizard.
• In (1) Choose a Source, specify the following:
  • For Enter a name, enter the physical schema name.
  • For Select a Datasource, select the Kafka external data source.
  • Optionally add a description.
• In the Schema Wizard footer, select Next.
• In (2) Manage Tables, in the Data Panel, select the topics you want to use to create tables. Select the Select All checkbox to select all the topics.
• In the Schema Wizard footer, select Next.
• In (3) Finalize, in the Schema Wizard footer, select Create Schema.

The Schema Wizard uses the default transformation query to discover Kafka messages and create related tables. You can edit the table’s dataset in the Table Editor and use a custom query.

Create a physical schema with the Schema Designer

Here are the steps to create a Kafka physical schema using the Schema Designer:

• Sign in to Incorta.
• In the Navigation bar, select Schema.
• In the action bar, select + New → Create Schema.
• In Name, specify the physical schema name, and select Save.
• In the Schema Designer, in the action bar, select + New → Table.
• In the Table Data Source dialog, specify the following:
  • For Type, select Streaming.
  • For Data Source, select the Kafka external data source.
  • Specify the Kafka table dataset properties.
• Select Add.
• In the Table Editor, in the Table Summary section, enter the table name.
• To save your changes, select Done in the action bar, and then in the Schema Designer, in the action bar, select Save Changes.

Kafka table dataset properties

For a physical schema table, you can define the following Kafka-specific dataset properties as follows:

View the physical schema diagram with the Schema Diagram Viewer

Here are the steps to view the physical schema diagram using the Schema Diagram Viewer:

• Sign in to Incorta.
• In the navigation bar, select Schema.
• In the list of physical schemas, select the Kafka physical schema.
• In the Schema Designer, in the summary section, select Diagram.

Load the physical schema

Here are the steps to perform a full load of the Kafka physical schema using the Schema Designer:

• Sign in to Incorta.
• In the navigation bar, select Schema.
• In the list of physical schemas, select the Kafka physical schema.
• In the Schema Designer, in the action bar, select Load → Full Load.
• To review the load status, in Last Load Status, select the date.

Here are the steps to perform an Incremental Load of the Kafka physical schema with the Schema Designer:

• Sign in to Incorta.
• In the navigation bar, select Schema.
• In the list of physical schemas, select the Kafka physical schema.
• In the Schema Designer, in the action bar, select Load → Incremental Load.
• To review the load status, in Last Load Status, select the date.

Explore the physical schema

With the full or incremental load of the Kafka physical schema complete, you can use the Analyzer to explore the physical schema, create your first insight, and save the insight to a new dashboard tab or an existing dashboard tab.

To open the Analyzer from the physical schema, follow these steps:

• In the navigation bar, select Schema.
• In the Schema Manager, in the physical schemas list, select the Kafka physical schema.
• In the Schema Designer, in the action bar, select Explore.

Additional Considerations

Data type discovery

The Kafka V2 connector may discover some data types as String during the JQ transformation. If the data types of your columns are not accurately discovered, you can adjust them manually by editing the dataset and selecting the correct data type in the Table Editor > Column Properties.

Transforming JSON messages

The Kafka V2 connector requires the JSON message to be valid JSON. By default, the connector discovers all top-level properties, objects, and arrays in the JSON messages as separate columns. However, nested (child) properties in arrays and objects are transformed into a single string. The Kafka V2 connector allows flattening the JSON message using JQ.

The following example shows the default transformation query of a Kafka message.

The Kafka message:

Copy
{  
   "store_id": 1,  
   "store_order_id": 1000,  
   "status": "accepted",  
   "branch": {  
       "branch_id": 5,  
       "state": "NC"  
   },  
   "order_lines": [  
       {"product_id": 10,  
           "category": "books",  
           "quantity": 1,  
           "unit_price": 15,  
           "net_price": 30},  
       {"product_id": 20,  
           "category": "notebooks",  
           "quantity": 5,  
           "unit_price": 12,  
           "net_price": 25}  
   ]}

The default transformation query:

Copy
{
    store_id: .store_id,
    store_order_id: .store_order_id,
    status: .status,
    branch: .branch,
order_lines: .order_lines
}

The output per the default transformation query:

Copy
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "branch": "{"branch_id": 5,"state": "NC"}",  
 "order_lines": "[{"product_id": 10,"category": "books","quantity": 1,"unit_price": 15,"net_price": 30},{"product_id": 20,"category": "notebooks","quantity": 5,"unit_price": 12,"net_price": 25}]"  
}

A customized transformation query:
You can edit the default transformation query and flatten objects and arrays to extract nested properties as separate columns or rows as required. You can use a JQ editor to flatten JSON objects and arrays.

Example 1:
The following is an example of a transformation query that flattens nested properties in an object to columns.

Copy
{
    store_id: .store_id,
    store_order_id: .store_order_id,
    status: .status,
    branch_id: .branch.branch_id,
    state: .branch.state,
}

The output per the customized transformation query in Example 1:

Copy
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "branch_id": 5,  
 "state": "NC"  
}

Example 2:
The following is an example of a transformation query that flattens nested properties in an array to rows.

Copy
{
    store_id: .store_id,
    store_order_id: .store_order_id,
    status: .status,
    order_lines: .order_lines[]
}  |
{
    store_id: .store_id,
    store_order_id: .store_order_id,
    status: .status,
product_id: .order_lines.product_id,
category: .order_lines.category,
quantity: .order_lines.quantity,
unit_price: .order_lines.unit_price,
net_price: .order_lines.net_price
}

The output per the customized transformation query in Example 2:

Copy
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "product_id": 10,  
 "category": "books",  
 "quantity": 1,  
 "unit_price": 15,  
 "net_price": 30  
}  
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "product_id": 20,  
 "category": "notebooks",  
 "quantity": 5,  
 "unit_price": 12,  
 "net_price": 25  
}

Example 3:
The following is an example of a transformation query that flattens nested properties in an object to columns and an array to rows.

Copy
.order_lines[] as $orderLine |
{
  store_id: .store_id,
  store_order_id: .store_order_id,
  status: .status,
  branch_id: .branch.branch_id,
  state: .branch.state,
  product_id: $orderLine.product_id,
  category: $orderLine.category,
  quantity: $orderLine.quantity,
  unit_price: $orderLine.unit_price,
  net_price: $orderLine.net_price
}

The output per the customized transformation query in Example 3:

Copy
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "branch_id": 5,  
 "state": "NC",  
 "product_id": 10,  
 "category": "books",  
 "quantity": 1,  
 "unit_price": 15,  
 "net_price": 30  
}  
{  
 "store_id": 1,  
 "store_order_id": 1000,  
 "status": "accepted",  
 "branch_id": 5,  
 "state": "NC",  
 "product_id": 20,  
 "category": "notebooks",  
 "quantity": 5,  
 "unit_price": 12,  
 "net_price": 25  
}