1.概述

　　大多数情况下，我们使用 Kafka 只是作为消息处理。在有些情况下，我们需要多次读取 Kafka 集群中的数据。当然，我们可以通过调用 Kafka 的 API 来完成，但是针对不同的业务需求，我们需要去编写不同的接口，在经过编译，打包，发布等一系列流程。最后才能看到我们预想的结果。那么，我们能不能有一种简便的方式去实现这一部分功能，通过编写 SQL 的方式，来可视化我们的结果。今天，笔者给大家分享一些心得，通过使用 SQL 的形式来完成这些需求。

2.内容

　　实现这些功能，其架构和思路并不复杂。这里笔者将整个实现流程，通过一个原理图来呈现。如下图所示：

　　这里笔者给大家详述一下上图的含义，消息数据源存放与 Kafka 集群当中，开启低阶和高阶两个消费线程，将消费的结果以 RPC 的方式共享出去（即：请求者）。数据共享出去后，回流经到 SQL 引擎处，将内存中的数据翻译成 SQL Tree，这里使用到了 Apache 的 Calcite 项目来承担这一部分工作。然后，我们通过 Thrift 协议来响应 Web Console 的 SQL 请求，最后将结果返回给前端，让其以图表的实行可视化。

3.插件配置

　　这里，我们需要遵循 Calcite 的 JSON Models，比如，针对 Kafka 集群，我们需要配置一下内容：

{
    version: '1.0',
    defaultSchema: 'kafka',  
    schemas: [  
        {
            name: 'kafka',  
            type: 'custom',
            factory: 'cn.smartloli.kafka.visual.engine.KafkaMemorySchemaFactory',  
            operand: {
                database: 'kafka_db'
            }  
        } 
    ]
}

　　另外，这里最好对表也做一个表述，配置内容如下所示：

[
    {
        "table":"Kafka",
        "schemas":{
            "_plat":"varchar",
            "_uid":"varchar",
            "_tm":"varchar",
            "ip":"varchar",
            "country":"varchar",
            "city":"varchar",
            "location":"jsonarray"
        }
    }
]

4.操作

　　下面，笔者给大家演示通过 SQL 来操作相关内容。相关截图如下所示：

　　在查询处，填写相关 SQL 查询语句。点击 Table 按钮，得到如下所示结果：

　　我们，可以将获取的结果以报表的形式进行导出。

　　当然，我们可以在 Profile 模块下，浏览查询历史记录和当前正在运行的查询任务。至于其他模块，都属于辅助功能（展示集群信息，Topic 的 Partition 信息等）这里就不多赘述了。

5.总结

　　分析下来，整体架构和实现的思路都不算太复杂，也不存在太大的难点，需要注意一些实现上的细节，比如消费 API 针对集群消息参数的调整，特别是低阶消费 API，尤为需要注意，其 fetch_size 的大小，以及 offset 是需要我们自己维护的。在使用 Calcite 作为 SQL 树时，我们要遵循其 JSON Model 和标准的 SQL 语法来操作数据源。

6.结束语

这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！