使用Lindorm流引擎处理Kafka数据 -

Lindorm流引擎提供了100%兼容Flink SQL的能力。您可以将原始数据存储在Kafka Topic，并通过Flink SQL在流引擎中创建实时计算任务，对原始数据进行高效计算和处理。本文介绍如何使用Flink SQL提交流引擎计算任务将Kafka Topic中的数据导入至Lindorm宽表。

前提条件

已开通Lindorm流引擎。如何开通，请参见开通流引擎。
已将客户端IP地址添加至Lindorm实例的白名单中。具体操作，请参见设置白名单。

注意事项

如果您的应用部署在ECS实例，且想要通过专有网络访问Lindorm实例，则需要确保Lindorm实例和ECS实例满足以下条件，以保证网络的连通性。

所在地域相同，并建议所在可用区相同（以减少网络延时）。
ECS实例与Lindorm实例属于同一专有网络。

操作步骤

步骤一：数据准备

通过Kafka API将源数据写入Kafka Topic。共支持以下两种写入方式：

以通过开源Kafka脚本工具写入数据为例。

#创建Topic
./kafka-topics.sh --bootstrap-server <Lindorm Stream Kafka地址> --topic log_topic --create

#写入数据
./kafka-console-producer.sh --bootstrap-server <Lindorm Stream Kafka地址> --topic log_topic
{"loglevel": "INFO", "thread":"thread-1", "class": "com.alibaba.stream.test", "detail":"thread-1 info detail", "timestamp": "1675840911549"}
{"loglevel": "ERROR", "thread":"thread-2", "class": "com.alibaba.stream.test", "detail":"thread-2 error detail", "timestamp": "1675840911549"}
{"loglevel": "WARN", "thread":"thread-3", "class": "com.alibaba.stream.test", "detail":"thread-3 warn detail", "timestamp": "1675840911549"}
{"loglevel": "ERROR", "thread":"thread-4", "class": "com.alibaba.stream.test", "detail":"thread-4 error detail", "timestamp": "1675840911549"}

Lindorm Stream Kafka地址的获取方式请参见查看连接地址。

在宽表引擎中创建结果表，用于存储计算任务的处理结果。
1. 通过Lindorm-cli连接宽表引擎。如何连接，请参见通过Lindorm-cli连接并使用宽表引擎。
2. 创建结果表log。
```
CREATE TABLE IF NOT EXISTS log (
  loglevel VARCHAR,
  thread VARCHAR,
  class VARCHAR,
  detail VARCHAR,
  timestamp BIGINT,
primary key (loglevel, thread) );
```

步骤二：安装流引擎客户端

在ECS上执行以下命令，下载流引擎客户端压缩包。

wget https://hbaseuepublic.oss-cn-beijing.aliyuncs.com/lindorm-sqlline-2.0.2.tar.gz

执行以下命令，解压压缩包。
```
tar zxvf lindorm-sqlline-2.0.2.tar.gz
```
进入lindorm-sqlline-2.0.2/bin目录，执行以下命令连接至Lindorm流引擎。
```
./lindorm-sqlline -url <Lindorm Stream SQL地址>
```
Lindorm Stream SQL地址的获取方式，请参见查看连接地址。

步骤三：在流引擎中提交计算任务

示例计算任务的具体实现如下：

创建名为log_to_lindorm的Flink Job，并在Flink Job中创建两张表：originalData（Source表）和lindorm_log_table（Sink表），分别关联已创建的Kafka Topic和结果表log。
创建流任务，过滤掉loglevel为ERROR的日志，将过滤结果写入结果表log中。

具体代码如下：

CREATE FJOB log_to_lindorm(
    --Kafka Source表
    CREATE TABLE originalData(
        `loglevel` VARCHAR,
        `thread` VARCHAR,
        `class` VARCHAR,
        `detail` VARCHAR,
        `timestamp` BIGINT
    )WITH(
        'connector'='kafka',
        'topic'='log_topic',
        'scan.startup.mode'='earliest-offset',
        'properties.bootstrap.servers'='Lindorm Stream Kafka地址',
        'format'='json'
    );
    --Lindorm宽表 Sink表
    CREATE TABLE lindorm_log_table(
        `loglevel` VARCHAR,
        `thread` VARCHAR,
        `class` VARCHAR,
        `detail` VARCHAR,
        `timestamp` BIGINT,
        PRIMARY KEY (`loglevel`, `thread`) NOT ENFORCED
    )WITH(
        'connector'='lindorm',
        'seedServer'='Lindorm宽表引擎的HBase兼容地址',
        'userName'='root',
        'password'='test',
        'tableName'='log',
        'namespace'='default'
    );
    --过滤Kafka中的ERROR日志，将结果写入Lindorm宽表
    INSERT INTO lindorm_log_table SELECT * FROM originalData WHERE loglevel = 'ERROR';
);

说明

Lindorm宽表引擎的HBase兼容地址的获取方式，请参见查看连接地址。
宽表连接器（Connector）的详细说明，请参见配置流引擎的宽表连接器。

步骤四：查询流引擎处理结果

支持以下两种查询方式：

通过Lindorm-cli连接宽表引擎并执行以下命令查询处理结果。

SELECT * FROM log LIMIT 5;

返回结果：

+----------+----------+-------------------------+-----------------------+---------------+
| loglevel |  thread  |          class          |        detail         |   timestamp   |
+----------+----------+-------------------------+-----------------------+---------------+
| ERROR    | thread-2 | com.alibaba.stream.test | thread-2 error detail | 1675840911549 |
| ERROR    | thread-4 | com.alibaba.stream.test | thread-4 error detail | 1675840911549 |
+----------+----------+-------------------------+-----------------------+---------------+

通过宽表引擎的集群管理系统查询处理结果，具体操作请参见数据查询。