1.概述
上次给大家分享了关于 Kafka SQL 的实现思路，这次给大家分享如何实现 Kafka SQL。要实现 Kafka SQL，在上一篇《Kafka - SQL 引擎分享》中分享了其实现的思路，核心包含数据源的加载，以及 SQL 树的映射。今天笔者给大家分享相关实现的代码。
2.内容
这里，将数据映射成 SQL Tree 是使用了 Apache Calcite 来承接这部分工作。在实现代码之前，我们首先来了解下 Apache Calcite 的相关内容，Apache Calcite 是一个面向 Hadoop 的查询引擎，它提供了业界标准的 SQL 语言，以及多种查询优化和连接各种存储介质的适配器。另外，还能处理 OLAP 和流处理场景。因为存在这么多优秀和闪光的特性， Hadoop 生态圈中 Apache Calcite 越发引人注目，被诸多项目所集成，常见的有：
- Apache Drill：基于大数据的实时查询引擎
- Apache Spark：继 Hadoop 之后的新一代大数据分布式处理框架。
- 更多详情，这里就不一一列举了，详情查看地址：《Adapters》
2.1 数据类型
这里数据源的数据类型，我们分为两种，一种是 SQL，另一种是基于编程语言的，这里我们所使用的是 Java，定义内容如下：
public static Map<String, SqlTypeName> SQLTYPE_MAPPING = new HashMap<String, SqlTypeName>(); public static Map<String, Class> JAVATYPE_MAPPING = new HashMap<String, Class>(); public static void initRowType() { SQLTYPE_MAPPING.put("char", SqlTypeName.CHAR); JAVATYPE_MAPPING.put("char", Character.class); SQLTYPE_MAPPING.put("varchar", SqlTypeName.VARCHAR); JAVATYPE_MAPPING.put("varchar", String.class); // ...... }
2.2 表的相关描述
另外，我们需要对表进行一个描述，在关系型数据库中，一个正常的表由行列组成，定义内容如下：
public static class Database { public List<Table> tables = new LinkedList<Table>(); } public static class Table { public String tableName; public List<Column> columns = new LinkedList<Column>(); public List<List<String>> data = new LinkedList<List<String>>(); } public static class Column { public String name; public String type; }
在每个集合中存储数据库相关名称，每个数据库存储多个集合的表对象，每个表对象下面又有一系列的列以及绑定的数据源。在每个列对象中包含字段名和类型，层层递进，依次关联。在使用 Calcite 是，需要遵循其 JSON Model，上篇博客我们已经定义过其 JSON Model，这里我们直接拿来使用，内容如下：
{ version: '1.0', defaultSchema: 'kafka', schemas: [ { name: 'kafka', type: 'custom', factory: 'cn.smartloli.kafka.visual.engine.KafkaMemorySchemaFactory', operand: { database: 'kafka_db' } } ] }
要实现其 Model ，这里需要我们去实现 org.apache.calcite.schema.SchemaFactory 的接口，内容如下所示：
public class KafkaMemorySchemaFactory implements SchemaFactory { @Override public Schema create(SchemaPlus parentSchema, String name, Map<String, Object> operand) { return new KafkaMemorySchema(name); } }
而在 KafkaMemorySchema 类中，我们只需要实现它的 getTableMap 方法，内容如下所示：
@Override protected Map<String, Table> getTableMap() { Map<String, Table> tables = new HashMap<String, Table>(); Database database = KafkaMemoryData.MAP.get(this.dbName); if (database == null) return tables; for (KafkaMemoryData.Table table : database.tables) { tables.put(table.tableName, new KafkaMemoryTable(table)); } return tables; }
从上述代码中，可以知道通过内存中的 Map 表查看对应的数据库对象，然后根据数据库对象中的表作为 Schema 中的表，而表的类型为 KafkaMemoryTable。
2.3 表类型
这里笔者就直接使用全表扫描，使用 org.apache.calcite.schema.impl.AbstractTable 的默认方式，实现其 getRowType 方法和 scan 方法，内容如下所示：
public RelDataType getRowType(RelDataTypeFactory typeFactory) { if(dataType == null) { RelDataTypeFactory.FieldInfoBuilder fieldInfo = typeFactory.builder(); for (KafkaMemoryData.Column column : this.sourceTable.columns) { RelDataType sqlType = typeFactory.createJavaType( KafkaMemoryData.JAVATYPE_MAPPING.get(column.type)); sqlType = SqlTypeUtil.addCharsetAndCollation(sqlType, typeFactory); fieldInfo.add(column.name, sqlType); } this.dataType = typeFactory.createStructType(fieldInfo); } return this.dataType; }
public Enumerable<Object[]> scan(DataContext root) { final List<String> types = new ArrayList<String>(sourceTable.columns.size()); for(KafkaMemoryData.Column column : sourceTable.columns) { types.add(column.type); } final int[] fields = identityList(this.dataType.getFieldCount()); return new AbstractEnumerable<Object[]>() { public Enumerator<Object[]> enumerator() { return new KafkaMemoryEnumerator<Object[]>(fields, types, sourceTable.data); } }; }
代码中，表中的字段名和类型是根据初始化时，每个表中的数据类型映射匹配的，在 KafkaMemoryData.SQLTYPE_MAPPING 和 KafkaMemoryData.JAVATYPE_MAPPING 中有描述相关自定义类型映射，这里就不多做赘述了。
实现流程大致就是这个样子，将每次的 SQL 查询，通过 Calcite 解析成标准可执行的 SQL 计划，执行期间会根据定义的信息，初始化每一个 Schema，在通过调用 getTableMap 获取字段名和类型，根据这些信息判断查询的表，字段名，类型以及 SQL 语法是否标准规范。然后在使用 Calcite 内部机制，生成物理执行计划。查询计划是 Tree 形式的，底层是进行扫表操作（可看作为 FROM），获取每个表的数据，之后在根据表数据进行上层的关联操作，如 JOIN，GROUP BY，LIMIT 等操作。
3.测试
完成上述流程后，进行代码测试，测试代码如下所示：
public static void main(String[] args) { try { Class.forName("org.apache.calcite.jdbc.Driver"); } catch (Exception ex) { ex.printStackTrace(); } Properties info = new Properties(); try { Connection connection = DriverManager.getConnection("jdbc:calcite:model=/Users/dengjie/hadoop/workspace/kafka/kafka-visual/src/main/resources/plugins.json",info); Statement st = connection.createStatement(); // String sql = "select * from \"Kafka\" where \"_plat\"='1004' limit 1"; String sql = "select * from \"Kafka\" limit 10"; long start = System.currentTimeMillis(); result = st.executeQuery(sql); ResultSetMetaData rsmd = result.getMetaData(); List<Map<String, Object>> ret = new ArrayList<Map<String,Object>>(); while (result.next()) { Map<String, Object> map = new HashMap<String, Object>(); for (int i = 1; i <= rsmd.getColumnCount(); i++) { System.out.print(result.getString(rsmd.getColumnName(i)) + " "); map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i))); } ret.add(map); System.out.println(); } System.out.println(new Gson().toJson(ret)); result.close(); connection.close(); } catch (SQLException e) { e.printStackTrace(); } }
4.总结
以上便是将 Kafka 中数据消费后，作为数据源加载和 SQL Tree 映射的实现代码，实现不算太困难，在编写 SQL 查询的时候，需要遵循标准的 SQL 语法来操作数据源。
5.结束语
这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！
邮箱：smartloli.org@gmail.com
Twitter： https://twitter.com/smartloli
QQ群（Hadoop - 交流社区1）： 424769183
温馨提示：请大家加群的时候写上加群理由（姓名＋公司/学校），方便管理员审核，谢谢！
