Kafka - SQL 代码实现

  1. 云栖社区>
  2. 博客列表>
  3. 正文

Kafka - SQL 代码实现

smartloli 2016-05-09 09:46:00 浏览179 评论0

摘要: 1.概述   上次给大家分享了关于 Kafka SQL 的实现思路,这次给大家分享如何实现 Kafka SQL。要实现 Kafka SQL,在上一篇《Kafka - SQL 引擎分享》中分享了其实现的思路,核心包含数据源的加载,以及 SQL 树的映射。

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
温馨提示:请大家加群的时候写上加群理由(姓名+公司/学校),方便管理员审核,谢谢!

热爱生活,享受编程,与君共勉!


作者:哥不是小萝莉 [关于我][犒赏

出处:http://www.cnblogs.com/smartloli/

转载请注明出处,谢谢合作!

【云栖快讯】你想见的Java技术专家都在这了,向大佬提问,有问题必答  详情请点击

网友评论

smartloli
文章175篇 | 关注112
关注
云数据库SQL Server,具有企业许可授权,权限更为开放,引擎功能更为强大。能完美支持W... 查看详情
云数据库PPAS版,是阿里云与EnterpriseDB公司合作基于PostgreSQL高度兼... 查看详情
PostgreSQL被业界誉为“最先进的开源数据库”,面向企业复杂SQL处理的OLTP在线事... 查看详情
为您提供简单高效、处理能力可弹性伸缩的计算服务,帮助您快速构建更稳定、安全的应用,提升运维效... 查看详情
阿里云总监课正式启航

阿里云总监课正式启航