lake: incremental updates 0621 #23112
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| --- | ||
| title: Kafka Consumer Integration Task (Beta) | ||
| summary: Create a Kafka Consumer task to continuously consume messages from Kafka topics and save the message content to internal object storage (tenant Stage). | ||
| --- | ||
|
|
||
| # Kafka Consumer Integration Task (Beta) | ||
|
|
||
| This page describes how to create a Kafka Consumer task that continuously consumes messages from Kafka topics and saves the message content to internal object storage (tenant Stage). | ||
|
|
||
| Unlike S3, MySQL, or PostgreSQL integration tasks, a Kafka Consumer task does not write directly to a regular target table. After the task is created and started, you can use the `@kafka_consumer/<task_name>/` stage path to view saved message objects and query their content with SQL. | ||
|
|
||
| If you need to create reusable Kafka connection settings first, see [Kafka - Credentials (Beta)](/tidb-cloud-lake/guides/kafka-credentials.md). | ||
|
|
||
| ## Use Cases | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| - Continuously ingest JSON messages from Kafka topics | ||
| - Land Kafka messages in internal object storage first, then query or process them with downstream SQL | ||
| - Preserve raw Kafka message objects for real-time or near-real-time data pipelines | ||
|
|
||
| ## Workflow | ||
|
|
||
| 1. An upstream system writes messages to Kafka topics. | ||
| 2. The Kafka Consumer task reads messages from the specified topics. | ||
| 3. The task saves messages in batches to internal object storage (tenant Stage). | ||
| 4. Users view generated objects through `@kafka_consumer/<task_name>/`. | ||
| 5. Users query message content from the stage and perform downstream loading or transformation as needed. | ||
|
|
||
| > **Note:** | ||
| > | ||
| > Kafka Consumer tasks save object files that contain Kafka message content. If you need to write messages into a business table, run downstream `INSERT INTO ... SELECT`, `COPY INTO`, or other processing based on the stage query results. | ||
|
|
||
| ## Prerequisites | ||
|
|
||
| Before creating a Kafka Consumer task, make sure: | ||
|
|
||
| - A **Kafka - Credentials** data source has already been created | ||
| - Platform can access the Kafka brokers over the network | ||
| - The authentication method, TLS settings, and account information in the Kafka data source are correct | ||
| - The Kafka user has permission to read the target topics | ||
| - Messages in the target topics match the **Data Format** selected in the task | ||
|
|
||
| ## Creating a Kafka Consumer Task | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| ### Step 1: Basic Info | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| 1. Navigate to **Data** > **Data Integration** and click **Create Task**. | ||
| 2. Select a Kafka data source, then configure the basic parameters: | ||
|
|
||
| | Field | Required | Description | | ||
| |-------|----------|-------------| | ||
| | **Data Source** | Yes | Select an existing **Kafka - Credentials** data source from the dropdown | | ||
| | **Name** | Yes | Name of the Kafka Consumer task | | ||
| | **Topics** | Yes | Kafka topics to consume. Separate multiple topics with commas, for example `topic-1,topic-2` | | ||
| | **Data Format** | Yes | Kafka message data format. Currently, this is **JSON** | | ||
| | **Start Position** | Yes | Start position when no committed offset exists. Supports **Latest** and **Earliest** | | ||
| | **Max Batch Bytes** | No | Maximum data size per batch. The default value is **16 MiB** | | ||
| | **Max Batch Wait Interval** | No | Maximum wait time per batch. The default value is **1 Minute** | | ||
|
|
||
| > **Note:** | ||
| > | ||
| > **Latest** consumes only new messages, while **Earliest** starts from the earliest retained messages in Kafka. This setting applies only when the Consumer Group has no committed offset and does not reset existing offsets. | ||
|
|
||
| ### Step 2: Preview Data | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| After completing the basic settings, click **Next** to enter **Preview Data Info**. | ||
|
|
||
| The system attempts to read sample messages from the specified Kafka topics. If messages are available, the page displays 1 to 2 JSON messages so you can verify the topics, data format, and message structure. | ||
|
|
||
| If no previewable messages are available, the page displays **No sample data available**. You can still continue creating the task, but we recommend checking whether the topics already contain messages and whether the selected **Start Position** can read sample data. | ||
|
|
||
| ### Step 3: Result Viewing | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| In the **Result Viewing** step, select the **Warehouse** to run the Kafka Consumer task. | ||
|
|
||
| After the task starts, it reads Kafka messages and saves them to internal object storage (tenant Stage). The page provides SQL examples. You can use `LIST @kafka_consumer/<task_name>/` to view generated objects and use stage queries to read message content. | ||
|
|
||
| ```sql | ||
| -- List stage objects: | ||
| LIST @kafka_consumer/<task_name>/; | ||
|
|
||
| -- Query object data (replace with the correct PATTERN path): | ||
| SELECT $1 | ||
| FROM @kafka_consumer ( | ||
| FILE_FORMAT=>'ndjson', | ||
| PATTERN=>'<task_name>/year=YYYY/month=MM/day=DD/hour=HH/.*[.]ndjson' | ||
| ); | ||
| ``` | ||
|
|
||
| Click **Create** to create the task. | ||
|
|
||
| ## Task Behavior | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| A Kafka Consumer task runs continuously. After it starts, it consumes messages from the specified topics and saves them in batches as object files in internal object storage until you stop it manually. | ||
|
|
||
| | Scenario | Behavior | | ||
| |----------|----------| | ||
| | New messages exist in the topics | Reads messages and writes them to the tenant Stage | | ||
| | The batch size reaches **Max Batch Bytes** | Writes the current batch to object storage | | ||
| | The wait time reaches **Max Batch Wait Interval** | Writes the current batch to object storage even if the batch does not reach the size limit | | ||
| | The write operation succeeds | Saves the consumption progress for later continuation | | ||
| | You stop the task manually | Stops consuming and keeps the saved message objects | | ||
|
|
||
| ## Query Saved Messages | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| Kafka Consumer tasks save message objects under the `@kafka_consumer/<task_name>/` path. After the task starts and writes objects, open the task details page and switch to the **Data Browsing** tab to view the object count and object list by UTC hour. | ||
|
|
||
| You can also use SQL to list objects first, then query their content based on the actual path: | ||
|
|
||
| ```sql | ||
| LIST @kafka_consumer/<task_name>/; | ||
| ``` | ||
|
|
||
| ```sql | ||
| SELECT $1 | ||
| FROM @kafka_consumer ( | ||
| FILE_FORMAT=>'ndjson', | ||
| PATTERN=>'<task_name>/year=YYYY/month=MM/day=DD/hour=HH/.*[.]ndjson' | ||
| ); | ||
| ``` | ||
|
|
||
| If you need to write messages into a business table, continue with downstream transformation or loading based on the query result. | ||
|
|
||
| ## Advanced Configuration | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| ### Runtime Size | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| Kafka Consumer tasks support changing the runtime size. Before changing Runtime Size, stop the task, then open the edit page from the **Edit** menu, select an appropriate runtime size in the **Runtime Size** section, and save the change. After you restart the task, it runs with the new runtime size. | ||
|
lilin90 marked this conversation as resolved.
|
||
|
|
||
| > **Note:** | ||
| > | ||
| > The available runtime sizes and prices depend on your billing plan. Use the options shown in the console and the pricing documentation as the source of truth. | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,43 @@ | ||||||
| --- | ||||||
| title: Kafka - Credentials (Beta) | ||||||
| summary: Create a "Kafka - Credentials" data source to store Kafka connection information for reuse in Kafka Consumer integration tasks. | ||||||
| --- | ||||||
|
|
||||||
| # Kafka - Credentials (Beta) | ||||||
|
|
||||||
| This page describes how to create a `Kafka - Credentials` data source. This data source stores the broker addresses, authentication method, and connection credentials required to access a Kafka cluster. You can reuse these settings across multiple Kafka Consumer integration tasks. | ||||||
|
|
||||||
| `Kafka - Credentials` only stores Kafka connection information. It does not consume messages by itself. The actual process of reading Kafka topic messages and writing them to internal object storage is performed by a [Kafka Consumer Integration Task (Beta)](/tidb-cloud-lake/guides/integrate-with-kafka.md). | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| ## Use Cases | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| - Centrally manage Kafka broker addresses and authentication settings | ||||||
| - Reuse the same Kafka connection settings across multiple Kafka Consumer tasks | ||||||
| - Update the Kafka addresses, authentication method, or account information in one place when referenced by multiple tasks | ||||||
|
Contributor
There was a problem hiding this comment. Avoid passive voice by rewriting the clause in active voice.
Suggested change
|
||||||
|
|
||||||
| ## Create Kafka - Credentials | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| 1. Navigate to **Data** > **Data Sources**, then click **Create Data Source**. | ||||||
| 2. Select **Kafka - Credentials** as the service type, then fill in the connection details: | ||||||
|
|
||||||
| | Field | Required | Description | | ||||||
| |-------|----------|-------------| | ||||||
| | **Name** | Yes | A descriptive name for the data source | | ||||||
| | **Brokers** | Yes | Kafka broker address list. Separate multiple addresses with commas, for example `broker-1:9092,broker-2:9093,broker-3:9092` | | ||||||
| | **Authentication** | Yes | Kafka authentication method. Supported options are **None** and **SASL/PLAIN** | | ||||||
| | **TLS encryption** | No | Whether to enable TLS encryption | | ||||||
| | **Username** | Required if applicable | Kafka username. Required when **SASL/PLAIN** is selected | | ||||||
| | **Password** | Required if applicable | Kafka password. Required when **SASL/PLAIN** is selected | | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| 3. Click **Test Connectivity** to validate the connection. If the test succeeds, click **OK** to save the data source. | ||||||
|
|
||||||
| ## Configuration Recommendations | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| - Create a dedicated Kafka user for the platform instead of sharing an application account. | ||||||
| - Enable **TLS encryption** if your Kafka cluster requires encrypted connections. | ||||||
| - If you select **SASL/PLAIN**, make sure the Kafka user is allowed to read the topics that will be consumed by downstream tasks. | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
| - Run **Test Connectivity** before saving the data source to verify the broker addresses, network access, and authentication settings. | ||||||
|
|
||||||
| ## Next Steps | ||||||
|
lilin90 marked this conversation as resolved.
|
||||||
|
|
||||||
| After creating the data source, you can use it to create a [Kafka Consumer Integration Task (Beta)](/tidb-cloud-lake/guides/integrate-with-kafka.md). | ||||||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.