Confluent Platform: ksqlDB 基本语法

女爷i 2022-10-09 02:45 136阅读 0赞

### 文章目录 ###

*  1, 资源管理/查看: show, describe, drop
 *  2, 集合管理: stream, table
 *   *  a, 创建stream/table: create xx
     *  b, 查看资源: show xx / describe xx
     *  c, 查询/插入数据 ( stream/table ): insert into / select xx from

ksql 语法说明doc： [https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/][https_docs.ksqldb.io_en_latest_developer-guide_ksqldb-reference]

# 1, 资源管理/查看: show, describe, drop #

*  在ksql中，非字母数字的资源名称要用\``反斜线`\`来表示 ( \``JdbcSourceConnector_mysql-t1`\` )

<table> 
 <thead> 
  <tr> 
   <th align="left">类别</th> 
   <th align="left">语法</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">查看资源类别</td> 
   <td align="left">show queries /topics /connectors /streams /tables ;</td> 
  </tr> 
  <tr> 
   <td align="left">查看stream/table详情</td> 
   <td align="left">describe [extended] table1;</td> 
  </tr> 
  <tr> 
   <td align="left">查看connector状态和错误信息</td> 
   <td align="left">describe connector `datagen-users` ;</td> 
  </tr> 
  <tr> 
   <td align="left">修改表结构</td> 
   <td align="left">alter table titles ADD COLUMN genre varchar ;</td> 
  </tr> 
  <tr> 
   <td align="left">删除资源</td> 
   <td align="left">drop { stream /table /connector} [ if exists ] [DELETE TOPIC];</td> 
  </tr> 
 </tbody> 
</table>

# 2, 集合管理: stream, table #

概念说明(stream/table)：[https://docs.ksqldb.io/en/latest/concepts/streams/][https_docs.ksqldb.io_en_latest_concepts_streams]

*  we don’t query Kafka topics directly. We query collections (i.e., streams and tables)
 *  stream ==> \[a=1, b=1, a=2\]， table ==> \{a=2, b=1\}  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

## a, 创建stream/table: create xx ##

*  Unlike tables, streams do not have a primary key column. Streams have immuta‐  
    ble, insert-style semantics, so uniquely identifying records is not possible. How‐  
    ever, the KEY identifier can be used to alias the record key column (which  
    corresponds to the Kafka record key

ksql> 
    //自定义数据类型( 结构体)
    CREATE TYPE season_length AS STRUCT<season_id INT, episode_count INT> ;
    //创建stream: 指定kafka topic (可以自动创建topic )
    CREATE STREAM production_changes (
     rowkey VARCHAR KEY, 
     uuid INT,
     title_id INT,
     change_type VARCHAR,
     before season_length, 
     after season_length,
     created_at VARCHAR
    ) WITH (
     KAFKA_TOPIC='production_changes',
     PARTITIONS='4',
     VALUE_FORMAT='JSON',
     TIMESTAMP='created_at', 
     TIMESTAMP_FORMAT='yyyy-MM-dd HH:mm:ss'
    );
    
    CREATE TABLE titles (
     id INT PRIMARY KEY,
     title VARCHAR
    ) WITH (
     KAFKA_TOPIC='titles',
     VALUE_FORMAT='AVRO',
     PARTITIONS=4
    );
    ALTER  table titles ADD COLUMN genre VARCHAR;

## b, 查看资源: show xx / describe xx ##

ksql> show streams;
     Stream Name         | Kafka Topic                 | Key Format | Value Format | Windowed
    ------------------------------------------------------------------------------------------
     PRODUCTION_CHANGES  | production_changes          | KAFKA      | JSON         | false
    ------------------------------------------------------------------------------------------
    
    ksql> show tables;
     Table Name | Kafka Topic | Key Format | Value Format | Windowed
    -----------------------------------------------------------------
     TITLES     | titles      | KAFKA      | AVRO         | false
    -----------------------------------------------------------------
    
    ksql> show topics;
     Kafka Topic                 | Partitions | Partition Replicas
    ---------------------------------------------------------------
     production_changes          | 4          | 1
     titles                      | 4          | 1
    ---------------------------------------------------------------
    
    ksql> describe titles extended;
    Name                 : TITLES
    Type                 : TABLE
    Timestamp field      : Not set - using <ROWTIME>
    Key format           : KAFKA
    Value format         : AVRO
    Kafka topic          : titles (partitions: 4, replication: 1)
    Statement            : CREATE TABLE TITLES (ID INTEGER PRIMARY KEY, TITLE STRING) WITH (KAFKA_TOPIC='titles', KEY_FORMAT='KAFKA', PARTITIONS=4, VALUE_FORMAT='AVRO');
     Field | Type
    ----------------------------------------
     ID    | INTEGER          (primary key)
     TITLE | VARCHAR(STRING)
    ----------------------------------------
    Local runtime statistics
    ------------------------
    (Statistics of the local KSQL server interaction with the Kafka topic titles

## c, 查询/插入数据 ( stream/table ): insert into / select xx from ##

*  query 类别 （push /pull )：[https://docs.ksqldb.io/en/latest/concepts/queries/][https_docs.ksqldb.io_en_latest_concepts_queries]
 *  sql语法： [push-query][]，[pull-query][]

// 插入数据
    ksql> 
    INSERT INTO titles VALUES (1, 'Stranger Things');
    
    INSERT INTO production_changes (
     ROWKEY,ROWTIME,
     uuid, title_id, change_type,
     before, after,
     created_at
    ) VALUES ('2',1581161400000,
     2, 2,'release_date',
     STRUCT(season_id := 1, release_date := '2021-05-27'),
     STRUCT(season_id := 1, release_date := '2021-08-18'),
     '2021-02-08 10:00:00'
    );
    
    // 查询数据
    		SELECT select_expr [, ...]
    		FROM from_item
    		[ LEFT JOIN join_collection ON join_criteria ]
    		[ WINDOW window_expression ]
    		[ WHERE condition ]
    		[ GROUP BY grouping_expression ]
    		[ PARTITION BY partitioning_expression ]
    		[ HAVING having_expression ]
    		EMIT CHANGES
    		[ LIMIT count ];
    ksql> set 'auto.offset.reset'='earliest';
    Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
    ksql> select * from production_changes emit changes limit 1;
    +--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
    |ROWKEY              |UUID                |TITLE_ID            |CHANGE_TYPE         |BEFORE              |AFTER               |CREATED_AT          |
    +--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
    |2                   |2                   |2                   |release_date        |{SEASON_ID=1, EPISOD|{SEASON_ID=1, EPISOD|2021-02-08 10:00:00 |
    |                    |                    |                    |                    |E_COUNT=null}       |E_COUNT=null}       |                    |
    Limit Reached
    Query terminated
    
    //1, like 模糊查询
    SELECT title_id, before, after, created_at
    FROM production_changes
    WHERE change_type LIKE 'releas%' 
    EMIT CHANGES ;
    
    //2, and, or: 多条件查询
    //CREATE TYPE season_length AS STRUCT<season_id INT, episode_count INT> ;
    //{SEASON_ID=1, EPISODE_COUNT=8}
    SELECT title_id, before, after, created_at
    FROM production_changes
    WHERE NOT change_type = 'release_date' 
    AND ( after->episode_count >= 8 OR after->episode_count <=20 ) 
    EMIT CHANGES ;
    
    //3, between x and y: 范围查询
    SELECT title_id, before, after, created_at
    FROM production_changes
    WHERE change_type = 'season_length'
    AND after->episode_count BETWEEN 8 AND 20
    EMIT CHANGES ;
    
    //4, coalesce, ifnull, case when: 条件非空判断
    // In this case, if after->season_id is NULL, then we fallback to before->season_id. 
    // If before->season_id is also NULL, then we will fallback to a default value of 0.
    SELECT COALESCE(after->season_id, before->season_id, 0) AS season_id
    FROM production_changes
    WHERE change_type = 'season_length'
    EMIT CHANGES ;
    
    SELECT IFNULL(after->season_id, before->season_id) AS season_id
    FROM production_changes
    WHERE change_type = 'season_length'
    EMIT CHANGES ;
    
    SELECT
     CASE
     WHEN after->season_id IS NOT NULL THEN after->season_id
     WHEN before->season;
     
    //5, cast 类型转换
    SELECT ... 
    FROM season_length_changes s
    INNER JOIN titles t
    ON CAST(s.title_id AS INT) = t.id 
    EMIT CHANGES ;

[https_docs.ksqldb.io_en_latest_developer-guide_ksqldb-reference]: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/
[https_docs.ksqldb.io_en_latest_concepts_streams]: https://docs.ksqldb.io/en/latest/concepts/streams/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20221005/75fa3055a1714e72a110343aa54cd3ef.png
[https_docs.ksqldb.io_en_latest_concepts_queries]: https://docs.ksqldb.io/en/latest/concepts/queries/
[push-query]: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/select-push-query/
[pull-query]: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/select-pull-query/