kafka早期作为一个日志消息系统,很受运维欢迎的,配合ELK玩起来很happy,在kafka慢慢的转向流式平台的过程中,开发也慢慢介入了,一些业务系统也开始和kafka对接起来了,也还是很受大家欢迎的,由于业务需要,一部分小白也就免不了接触kafka了,这些小白总是会安奈不住好奇心,要精确的查看kafka中的某一条数据,作为服务提供方,我也很方啊,该怎么怼?业务方不敢得罪啊,只能写consumer去消费,然后人肉查询。

    有什么方法能直接查询kafka中已有的数据呢?那时候presto就映入眼帘了,初步探索后发现presto确实强大,和我们在用的impala有的一拼,支持的数据源也更多,什么redis、mongo、kafka都可以用sql来查询,真是救星啊,这样那群小白就可以直接使用presto来查询里面的数据了。不过presto在不开发插件的情况下,对kafka的数据有格式要求,支持json、avro。关于presto的调研见presto实战。但是我只是想用sql查询kafka,而presto功能过于强大,必然整个框架就显得比较厚重了,功能多嘛。有什么轻量级的工具呢?

    某一天,kafka的亲儿子KSQL就诞生了,KSQL是一个用于Apache kafka的流式SQL引擎,KSQL降低了进入流处理的门槛,提供了一个简单的、完全交互式的SQL接口,用于处理Kafka的数据,可以让我们在流数据上持续执行 SQL 查询,KSQL支持广泛的强大的流处理操作,包括聚合、连接、窗口、会话等等。

    KSQL在内部使用Kafka的Streams API,并且它们共享与Kafka流处理相同的核心抽象,KSQL有两个核心抽象,它们对应于到Kafka Streams中的两个核心抽象,让你可以处理kafka的topic数据。关于这两个核心抽象下章节解读。

        由一个KSQL服务器进程执行查询。一组KSQL进程可以作为集群运行。可以通过启动更多的KSQL实例来动态添加更多的处理能力。这些KSQL实例是容错的,如果一个实例失败了,其他的就会接管它的工作。查询是使用交互式的KSQL命令行客户端启动的,该客户端通过REST API向集群发送命令。命令行允许检查可用的stream和table,发出新的查询,检查状态并终止正在运行的查询。KSQL内部是使用Kafka的stream API构建的,它继承了它的弹性可伸缩性、先进的状态管理和容错功能,并支持Kafka最近引入的一次性处理语义。KSQL服务器将此嵌入到一个分布式SQL引擎中(包括一些用于查询性能的自动字节代码生成)和一个用于查询和控制的REST API。

    KSQL简化了流应用程序,它集成了stream和table的概念,允许使用表示现在发生的事件的stream来连接表示当前状态的table。 Apache Kafka中的一个topic可以表示为KSQL中的STREAM或TABLE,具体取决于topic处理的预期语义。下面看看两个核心的解读。

        stream:流是无限制的结构化数据序列,stream中的fact是不可变的,这意味着可以将新fact插入到stream中,但是现有fact永远不会被更新或删除。 stream可以从Kafka topic创建,或者从现有的stream和table中派生。

        table:一个table是一个stream或另一个table的视图,它代表了一个不断变化的fact的集合,它相当于传统的数据库表,但通过流化等流语义来丰富。表中的事实是可变的,这意味着可以将新的事实插入到表中,现有的事实可以被更新或删除。可以从Kafka主题中创建表,也可以从现有的流和表中派生表。

    ksql支持kafka0.11之后的版本,在confluent的V3和V4版本中默认并没有加入ksql server程序,当然V3和V4是支持ksql的,在V5版本中已经默认加入ksql了,为了方便演示,我们使用confluent kafka V5版本演示,zk和kafka也是单实例启动。

  1. wget https://packages.confluent.io/archive/5.0/confluent-oss-5.0.0-2.11.tar.gz
  2. tar zxvf confluent-oss-5.0.0-2.11.tar.gz -C /opt/programs/confluent_5.0.0
  1. cd /opt/programs/confluent_5.0.0
  2. bin/zookeeper-server-start -daemon etc/kafka/zookeeper.properties
  1. cd /opt/programs/confluent_5.0.0
  2. bin/kafka-server-start -daemon etc/kafka/server.properties

        confluent自带了一个ksql-datagen工具,可以创建和产生相关的topic和数据,ksql-datagen可以指定的参数如下:

  1. [bootstrap-server=<kafka bootstrap server(s)> (defaults to localhost:9092)]
  2. [quickstart=<quickstart preset> (case-insensitive; one of \'orders\', \'users\', or \'pageviews\')]
  3. schema=<avro schema file>
  4. [schemaRegistryUrl=<url for Confluent Schema Registry> (defaults to http://localhost:8081)]
  5. format=<message format> (case-insensitive; one of \'avro\', \'json\', or \'delimited\')
  6. topic=<kafka topic name>
  7. key=<name of key column>
  8. [iterations=<number of rows> (defaults to 1,000,000)]
  9. [maxInterval=<Max time in ms between rows> (defaults to 500)]
  10. [propertiesFile=<file specifying Kafka client properties>]

        创建pageviews,数据格式为delimited

  1. cd /opt/programs/confluent_5.0.0/bin
  2. ./ksql-datagen quickstart=pageviews format=delimited topic=pageviews maxInterval=500

            ps:以上命令会源源不断在stdin上输出数据,就是工具自己产生的数据,如下样例

  1. 8001 --> ([ 1539063767860 | \'User_6\' | \'Page_77\' ]) ts:1539063767860
  2. 8011 --> ([ 1539063767981 | \'User_9\' | \'Page_75\' ]) ts:1539063767981
  3. 8021 --> ([ 1539063768086 | \'User_5\' | \'Page_16\' ]) ts:1539063768086

            不过使用consumer消费出来的数据是如下样式

  1. 1539066430530,User_5,Page_29
  2. 1539066430915,User_6,Page_74
  3. 1539066431192,User_4,Page_28
  4. 1539066431621,User_6,Page_38
  5. 1539066431772,User_7,Page_29
  6. 1539066432122,User_8,Page_34

        创建users,数据格式为json

  1. cd /opt/programs/confluent_5.0.0/bin
  2. ./ksql-datagen quickstart=users format=json topic=users maxInterval=100

        ps:以上命令会源源不断在stdin上输出数据,就是工具自己产生的数据,如下样例

  1. User_5 --> ([ 1517896551436 | \'User_5\' | \'Region_5\' | \'MALE\' ]) ts:1539063787413
  2. User_7 --> ([ 1513998830510 | \'User_7\' | \'Region_4\' | \'MALE\' ]) ts:1539063787430
  3. User_6 --> ([ 1514865642822 | \'User_6\' | \'Region_2\' | \'MALE\' ]) ts:1539063787481

        不过使用consumer消费出来的数据是如下样式

  1. {"registertime":1507118206666,"userid":"User_6","regionid":"Region_7","gender":"OTHER"}
  2. {"registertime":1506192314325,"userid":"User_1","regionid":"Region_1","gender":"MALE"}
  3. {"registertime":1489277749526,"userid":"User_6","regionid":"Region_4","gender":"FEMALE"}
  4. {"registertime":1497188917765,"userid":"User_9","regionid":"Region_3","gender":"OTHER"}
  5. {"registertime":1493121964253,"userid":"User_4","regionid":"Region_3","gender":"MALE"}
  6. {"registertime":1515609444511,"userid":"User_5","regionid":"Region_9","gender":"FEMALE"}
  1. cd /opt/programs/confluent_5.0.0
  2. bin/ksql-server-start -daemon etc/ksql/ksql-server.properties
  1. cd /opt/programs/confluent_5.0.0
  2. bin/ksql http://10.205.151.145:8088

   

            根据topic pageviews创建一个stream pageviews_original,value_format为DELIMITED

  1. ksql>CREATE STREAM pageviews_original (viewtime bigint, userid varchar, pageid varchar) WITH \
  2. (kafka_topic=\'pageviews\', value_format=\'DELIMITED\');

            根据topic users创建一个table users_original,value_format为json

  1. ksql>CREATE TABLE users_original (registertime BIGINT, gender VARCHAR, regionid VARCHAR, userid VARCHAR) WITH \
  2. (kafka_topic=\'users\', value_format=\'JSON\', key = \'userid\');

  1. ksql> SELECT * FROM USERS_ORIGINAL LIMIT 3;
  2. ksql> SELECT * FROM pageviews_original LIMIT 3;

        ps:ksql默认是从kafka最新的数据查询消费的,如果你想从开头查询,则需要在会话上进行设置:SET \’auto.offset.reset\’ = \’earliest\’;

        持久化查询可以源源不断的把查询出的数据发送到你指定的topic中去,查询的时候在select前面添加create stream关键字即可创建持久化查询。

  1. ksql> CREATE STREAM pageviews2 AS SELECT userid FROM pageviews_original;

  1. ksql> SHOW STREAMS;

            ps:可以看到新创建了stream PAGEVIEWS2,并且创建了topic PAGEVIEWS2

  1. ksql> SHOW QUERIES;

            ps:可以看到ID为CSAS_PAGEVIEWS2_0的任务在执行,并且有显示执行的语句

  1. cd /opt/programs/confluent_5.0.0/bin
  2. ./kafka-console-consumer --bootstrap-server 10.205.151.145:9092 --from-beginning --topic
  3. PAGEVIEWS2

            ps:可以看到PAGEVIEWS2 topic里面正是我们通过select筛选出来的数据

  1. ksql> TERMINATE CSAS_PAGEVIEWS2_0;

版权声明:本文为xibuhaohao原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/xibuhaohao/p/11490675.html