This article describes how to integrate Kafka to create datasets in order to support real-time data analytics scenarios for products.

The Kafka data source is only supported in private deployment mode. To use the Kafka data source in the SaaS version, please contact your company's customer success manager to submit your requirements.

1. Log in to Byteplus, navigate to a specific project, click Data Preparation, find Data Connection in the dropdown list, and click Data Connection.

   

2. Select Kafka on the page.

3. Fill in the required basic information, test the connection, and click Save after the connection is successful.
   
   Kafka provides four authentication methods. The parameter configuration instructions for each authentication method are as follows. Users need to select according to their actual scenario:

   Authentication methods	Note
   No authentication	Extremely low security, suitable for development, testing, or local environments. Data connection name: The name of the new data connection, used to distinguish different database connections in the system. The default is Kafka_timestamp. You can modify the name, but please enter a standard name. Only combinations of Chinese characters, English letters, digits, and underscores are allowed, and spaces are not permitted at the beginning or end. Version: Specify the Kafka version for the connection. Server: The server address and port number in the format ip:port. For multiple entries, separate them with commas.
   SASL_PLAINTEXT	Low security (authentication only), suitable for internal trusted networks and lightweight authentication. Data connection name: The name of the new data connection, used to distinguish different database connections in the system. The default is Kafka_timestamp. You can modify the name, but please enter a standard name. Only combinations of Chinese characters, English letters, digits, and underscores are allowed, and spaces are not permitted at the beginning or end. Username: SASL username. Password: SASL password. Version: Specify the Kafka version for the connection. Server: The server address and port number in the format ip:port. For multiple entries, separate them with commas. Encryption method: The encryption method for the connection. Depending on the SASL user type, you can choose from three methods: PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.
   Built-in Kerberos for the cluster	High security, suitable for enterprise production environments and multi-tenant clusters. Data connection name: The name of the new data connection, used to distinguish different database connections in the system. The default is Kafka_timestamp. You can modify the name, but please enter a standard name. Only combinations of Chinese characters, English letters, digits, and underscores are allowed, and spaces are not permitted at the beginning or end. Version: Specify the Kafka version for the connection. Server: The server address and port number in the format ip:port. For multiple entries, separate them with commas.
   Custom Kerberos	High security (scalable), suitable for cross-domain authentication and integration with existing Kerberos architectures. Data connection name: The name of the new data connection, used to distinguish different database connections in the system. The default is Kafka_timestamp. You can modify the name, but please enter a standard name. Only combinations of Chinese characters, English letters, digits, and underscores are allowed, and spaces are not permitted at the beginning or end. Version: Specify the Kafka version for the connection. Server: The server address and port number, formatted as ip:port. Use commas to separate multiple entries. Krb5 file: The configuration file for Kerberos, which contains information such as the Kerberos server address and realm (Realm). The client requires this file to interact with the Kerberos server and obtain authentication credentials such as tickets (Ticket). Keytab file: Contains the key information for the client or server, used for identity verification during Kerberos authentication. serviceName: The name of the Kafka service in Kerberos authentication. This name must correspond to the service name configured on the Kafka server and is used to register and verify the service identity on the Kerberos server. Principal: Represents the identity of the client or server within the Kerberos realm, typically as a string in the format {username}/{hostname}@{REALM}. During authentication, the Kerberos server verifies the legitimacy of the identity based on the Principal.

1. Log in to Byteplus, click Data Preparation --> Dataset, and click the "New" button in the upper left corner to create a new dataset.

   

2. When selecting the data connection, click New Configuration.

   

3. The subsequent steps are the same as steps 2 and 3 in Create from data connection. After completing the configuration, you will remain in the popup window to select the data connection for the dataset, where you can proceed directly to the next step of dataset creation.

1. Use the previously created data connection to create a dataset: Click Data Preparation --> Dataset, and click the "New" button in the upper left corner to create a new dataset.

   

2. Search for or select from the dropdown the previously created data connection.

3. Select the required fields and their corresponding data types.

   warning

   • Writing to table after aggregation is not supported, and overwriting historical data is not supported.
   • Kafka topics do not currently support joins; only single-table dataset creation is supported.
   • When the topic data volume is too small, please manually enter JSON extraction.

   • The configuration supports nested JSON. You need to use jsonpath to extract fields. For details, see Explanation of nested field extraction rules.
     Example: outter.inner.cnt means to obtain 0 from {"outter": {"inner": {"cnt": 0}}}.

     

     warning

     Field names starting with an underscore are not supported, and fields containing the '-' character are not supported.

   • Kafka dataset data type mapping. For the rules of Kafka partition keys, see Explanation of the rules for partition keys.

Since the Kafka data source integration relies on ClickHouse at the underlying layer, when creating a dataset, you need to select the corresponding ClickHouse field type according to the Kafka data type (JSON) to ensure compatibility and processing efficiency of the data format.
  The following is the mapping relationship between the two, to guide you in selecting the correct mapped field type during field configuration:
  
  | | | \
  |JSON type |ClickHouse alternative types |
  |---|---|
  | | | \
  |long |Int64 |
  | | | \
  |string |String, Date, Datatime |
  | | | \
  |float |Float64 |
  | | | \
  |boolean |Int64 |
  | | | \
  |array |Array(Int64), Array(Float64), Array(String) |
  | | | \
  |map |Map(Int64, Int64), Map(Int64, Float64), Map(Int64, String) |\
  | |Map(String, Int64), Map(String, Float64), Map(String, String) |

Explanation of nested field extraction rules #

When there are multiple levels of nesting, you can enter a JSON path to extract the field.
Example:

{
  "body": {
    "cost": {
      "city": "Shanghai",
      "country": "China"
    }
  }
}

To extract the field city, set the json.path to body.cost.city.

Example operation path I

1. When creating a dataset, specify the field test as a non-nested field (without json.path), with the type Nullable(Int64).
2. To change test to a nested field (with json.path), delete the test field and click Save, then edit again, add the test field and json.path, and save.
   Example operation path II
3. When creating a dataset, specify the field test as a non-nested field (without json.path), with the type Nullable(Int64).
4. To change the type of test to Nullable(String), modify it directly and save.
   Example operation path III
5. When creating a dataset, specify the field test as a nested field (with json.path), with the type Nullable(Int64).
6. To change test to a non-nested field (without json.path), delete the test field, click Save, then edit again, add the test field and specify the type, and save.
   

Explanation of the rules for partition keys #

The Kafka partition key must be convertible to date (toDate) or datetime (toDateTime) format to support subsequent time-based operations on the partition key, such as partitioning by day or querying by time range.
Currently, Kafka only supports the following two formats:

• int type timestamp (supports seconds or milliseconds); it is recommended to use int type timestamp.
• String type date, in the format '2020-01-01' (date) or '2020-01-01 00:00:00' (datetime).

If you create a table via JSON, the value of the partition key in the JSON must also follow the above rules:

• int type timestamp: select field type Int64 (for example, 1672531200000 represents 2023-01-01 00:00:00).
• string type date (format 'YYYY-MM-DD'): select field type Date (for example, '2020-01-01').
• string type datetime (format 'YYYY-MM-DD HH:MM:SS'): select field type DateTime (for example, '2020-01-01 00:00:00').