How to access Kafka data using the Lindorm compute engine-Lindorm(Lindorm)-阿里云帮助中心

This topic describes how to access Kafka data using the Lindorm compute engine. You can load data from a Kafka instance into the Lindorm compute engine and combine it with data from other data sources to complete complex data production jobs.

Prerequisites

The compute engine service is enabled for your Lindorm instance. For more information, see Enable the service.
A Kafka instance is created in the same region as your Lindorm instance. For more information, see Internet and VPC connections.
A topic is created in the Kafka instance. For more information, see Step 3: Create resources.

Procedure

Start Beeline and access the Lindorm compute engine service. For more information, see Use Beeline to access the JDBC service.
Log on to the ApsaraMQ for Kafka console and obtain the connection information for accessing Kafka data.
1. On the Instance Details page, in the Endpoint Information section, obtain the domain endpoint of the Kafka instance.
2. On the Topic Management page, obtain the topic name of the Kafka instance.

Create a temporary Hive table in the Lindorm compute engine that corresponds to the Kafka topic.

CREATE TEMPORARY TABLE kafka_tbl USING kafka OPTIONS(table_options);

USING kafka specifies Kafka as the data source. A schema is not required for a Kafka table. The Lindorm compute engine provides a default Kafka schema that cannot be modified. The schema information is as follows:

Column name	Data type	Description
key	binary	The key of the record in Kafka.
value	binary	The value of the record in Kafka.
topic	string	The Kafka topic to which the record belongs.
partition	int	The partition to which the record belongs.
offset	bigint	The offset of the record in the partition.
timestamp	timestamp	The timestamp of the record.
timestampType	int	The type of the record's timestamp: 0: CreateTime. 1: LogAppendTime. For more information, see timestampType.

The following table describes the parameters in table_options.

Parameter	Required	Description	Example Value
kafka.bootstrap.servers	Yes	The domain endpoint of the Kafka instance.	alikafka-pre-cn-x0r34a20**-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20**-3-vpc.alikafka.aliyuncs.com:9092
subscribe	Yes	The topic name of the Kafka instance.	topic_to_read
startingTimestamp	Yes	The start timestamp of the data segment to access in the topic. Unit: milliseconds. Note Use the Spark SQL unix_timestamp function to convert time to a UNIX timestamp.	1682406000000
endingTimestamp	Yes	The end timestamp of the data segment to access in the topic. Unit: milliseconds. Note Use the Spark SQL unix_timestamp function to convert time to a UNIX timestamp.	1682409600000
startingOffsetsByTimestampStrategy	No	Add this parameter if some partitions in the Kafka instance contain no data.	latest

Creating a Kafka table supports various types of parameters. For more information about the parameters, see Structured Streaming + Kafka Integration Guide.

Example:

CREATE TEMPORARY TABLE kafka_tbl USING kafka OPTIONS(
  "kafka.bootstrap.servers"="alikafka-pre-cn-x0r34a20****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-3-vpc.alikafka.aliyuncs.com:9092",
  "subscribe"="topic_to_read",
  "startingTimestamp"=1682406000000,
  "endingTimestamp"=1682409600000);

You can query data from the Kafka table.
- You can query the schema data of the kafka_tbl table.
```
DESCRIBE kafka_tbl;
```
- You can query data from the kafka_tbl table.
```
SELECT * FROM kafka_tbl LIMIT 10;
```
- You can use a Spark function to fetch data from Kafka. For example, you can query data from the kafka_tbl table where the value is {"content": "kafka record"}.
```
SELECT get_json_object(cast(value as string), '$.content') FROM kafka_tbl LIMIT 10;
```
  The following result is returned:
```
Output: kafka record
```

(Optional) Example: Import Kafka data into a Hive table

To perform data analytics, you can import data from Kafka into a Hive table by following these steps.

Assume that a Kafka instance has the domain endpoint kafka_addr:9092 and a topic named topic1. The instance contains two data records that were written between 15:00:00 on 2023-04-25 and 16:00:00 on 2023-04-25. The content is as follows:

{"id": 1, "name": "name1"}
{"id": 2, "name": "name2"}

You will write these two records to a Hive table for data analytics.

Create a Kafka source table.

CREATE TEMPORARY TABLE kafka_src_tbl USING kafka OPTIONS(
  "kafka.bootstrap.servers"="kafka_addr:9092",
  "subscribe"="topic1",
  "startingTimestamp"=1682406000000,
  "endingTimestamp"=1682409600000);

For more information about the parameters, see Parameters.

Create a Hive destination table.

CREATE TABLE kafka_target_tbl(id LONG, name STRING) USING parquet;

Read all data from the Kafka source table and write it to the Hive destination table.

INSERT INTO kafka_target_tbl
    SELECT
      cast(get_json_object(cast(value as string), '$.id') as long),
      get_json_object(cast(value as string), '$.name')
    FROM
      kafka_src_tbl;

Query the Kafka data in a Hive table.

SELECT * FROM kafka_target_tbl LIMIT 10;

The following result is returned:

Output:
+-----+--------+
| id  |  name  |
+-----+--------+
| 1   | name1  |
| 2   | name2  |
+-----+--------+