Doris初步使用2

绝地灬酷狼 2022-11-27 08:57 379阅读 0赞

>   
> 使用命令查看创建表语句：  
> HELP CREATE TABLE;
> 
> 创建表完整语句：  
>     CREATE \[EXTERNAL\] TABLE \[IF NOT EXISTS\] \[database.\]table\_name  
>     (column\_definition1\[, column\_definition2, ...\]  
>     \[, index\_definition1\[, ndex\_definition12,\]\])  
>     \[ENGINE = \[olap|mysql|broker\]\]  
>     \[key\_desc\]  
>     \[COMMENT "table comment"\];  
>     \[partition\_desc\]  
>     \[distribution\_desc\]  
>     \[rollup\_index\]  
>     \[PROPERTIES ("key"="value", ...)\]  
>     \[BROKER PROPERTIES ("key"="value", ...)\]   
>    
> 1）创建表 在某些多维分析场景下，数据既没有主键，也没有聚合需求。因此，我们引入 Duplicate 数据模型来满足这类需求  
>     DUPLICATE KEY 是按字段排序：  
>       
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table  
> (  
>     aid INT NOT NULL COMMENT "文章id",  
>     title VARCHAR(32) COMMENT "标题",  
>     content VARCHAR(32) COMMENT "评论",  
>     uid VARCHAR(32) COMMENT "用户id",  
>     tid VARCHAR(32) COMMENT "文章类型"  
> )  
> DUPLICATE KEY(aid)  
> DISTRIBUTED BY HASH(aid) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
> \----在zeppelin里面执行 ：  
> %doris  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table2  
> (  
>     aid INT NOT NULL COMMENT "文章id",  
>     title VARCHAR(32) COMMENT "标题",  
>     content VARCHAR(32) COMMENT "评论",  
>     uid VARCHAR(32) COMMENT "用户id",  
>     tid VARCHAR(32) COMMENT "文章类型"  
> )  
> DUPLICATE KEY(aid)  
> DISTRIBUTED BY HASH(aid) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
> \---创建成功之后我们可以通过mysql 客户端查看到表跟信息。目前表是空的
> 
> \---我们需要导入数据，通过kafka实时导入，格式为json  
> \---具体的细节信息通过 HELP ROUTINE LOAD;
> 
> CREATE ROUTINE LOAD example\_db.duplicate\_insert2 ON duplicate\_table2  
>         COLUMNS(aid,title,content,uid,tid)  
>         PROPERTIES  
>         (  
>         "desired\_concurrent\_number"="3",  
>         "max\_batch\_interval" = "20",  
>         "max\_batch\_rows" = "300000",  
>         "max\_batch\_size" = "209715200",  
>         "strict\_mode" = "false",  
>         "format" = "json"  
>         )  
>            FROM KAFKA  
>         (  
>             "kafka\_broker\_list" = "192.168.12.188:9092",  
>             "kafka\_topic" = "doris\_article2",  
>             "kafka\_partitions" = "0",  
>             "kafka\_offsets" = "OFFSET\_BEGINNING"  
>         );
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]
> 
>  
> 
> \---执行语句，查看过程：  
> SHOW ROUTINE LOAD;
> 
> | 10096 | duplicate\_insert | 2020-08-14 11:49:50 | N/A       | N/A     | default\_cluster:example\_db | duplicate\_table | RUNNING | KAFKA          | 1              | \{"partitions":"\*","columnToColumnExpr":"aid,title,content,uid,tid","maxBatchIntervalS":"20","whereExpr":"\*","dataFormat":"json","maxBatchSizeBytes":"209715200","maxErrorNum":"0","currentTaskConcurrentNum":"1","maxBatchRows":"300000"\}   | \{"topic":"doris\_article","currentKafkaPartitions":"0","brokerList":"192.168.12.188:9092"\} | \{\}               | \{"receivedBytes":0,"errorRows":0,"committedTaskNum":0,"loadedRows":0,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":0,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":1\} 
> 
> \---查看mysql  
> 没数据 报错了：  
> STOP ROUTINE LOAD FOR duplicate\_insert;  杀掉  
>           
> \---具体的细节信息通过 HELP ROUTINE LOAD;  
>  ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]
> 
>   
> \----------之前正确案例
> 
> %doris  
>   CREATE TABLE example\_db.testing\_table  
> (  
>     siteid INT DEFAULT '10',  
>     username VARCHAR(32) DEFAULT '',  
>     pv BIGINT SUM DEFAULT '0'  
> )  
> AGGREGATE KEY(siteid, username)  
> DISTRIBUTED BY HASH(siteid) BUCKETS 10  
> PROPERTIES("replication\_num" = "1");
> 
> %doris  
> CREATE ROUTINE LOAD example\_db.test\_json\_label ON testing\_table  
>         COLUMNS(category,price,author)  
>         PROPERTIES  
>         (  
>         "desired\_concurrent\_number"="1",  
>         "max\_batch\_interval" = "20",  
>         "max\_batch\_rows" = "300000",  
>         "max\_batch\_size" = "209715200",  
>         "strict\_mode" = "false",  
>         "format" = "json"  
>         )  
>         FROM KAFKA  
>         (  
>         "kafka\_broker\_list" = "192.168.6.34:9092,192.168.6.35:9092,192.168.6.36:9092",  
>         "kafka\_topic" = "doris\_topic",  
>         "kafka\_partitions" = "0",  
>         "kafka\_offsets" = "0"  
>         );
> 
> \---再次测试  duplicate 模型  
>  
> 
>  ---案例写法  
>  CREATE TABLE example\_db.duplicate\_table  
> (  
>     \`siteid\` INT DEFAULT '10',  
>     \`username\` VARCHAR(32) DEFAULT '',  
>     \`error\_code\` INT COMMENT "错误码"  
> )  
> DUPLICATE KEY(\`siteid\`, \`username\`)  
> DISTRIBUTED BY HASH(siteid) BUCKETS 10  
> PROPERTIES("replication\_num" = "1");
> 
> CREATE ROUTINE LOAD example\_db.duplicate\_load ON duplicate\_table  
>     COLUMNS(\`siteid\`,\`username\`,\`error\_code\`)  
>     PROPERTIES  
>     (  
>     "desired\_concurrent\_number"="3",  
>     "max\_batch\_interval" = "20",  
>     "max\_batch\_rows" = "300000",  
>     "max\_batch\_size" = "209715200",  
>     "strict\_mode" = "false",  
>     "format" = "json"  
>     )  
>     FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "test",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
>     );
> 
>   
>   ---正常写法  
>  CREATE TABLE example\_db.duplicate\_table2  
> (  
>     siteid INT DEFAULT '10',  
>     username VARCHAR(32) DEFAULT '',  
>     error\_code INT COMMENT "错误码"  
> )  
> DUPLICATE KEY(siteid, username)  
> DISTRIBUTED BY HASH(siteid) BUCKETS 10  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.aaa ON duplicate\_table2  
> COLUMNS(siteid,username,error\_code)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "ods\_kafka",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
>     );  
>       
> \----3个表案例,这里使用的是DUPLICATE模型，存入是的原始数据：
> 
> \--article 表  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table\_01  
> (  
>     aid INT NOT NULL COMMENT "文章id",  
>     title VARCHAR(32) COMMENT "标题",  
>     content VARCHAR(32) COMMENT "评论",  
>     uid VARCHAR(32) COMMENT "用户id",  
>     tid VARCHAR(32) COMMENT "文章类型"  
> )  
> DUPLICATE KEY(aid,title)  
> DISTRIBUTED BY HASH(aid) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.article ON duplicate\_table\_01  
> COLUMNS(aid,title,content,uid,tid)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "doris\_article\_test1",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
>     );  
>       
>       
>       
> \--user 表  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table\_02  
> (  
>        
>     uid VARCHAR(32) COMMENT "用户id",  
>     username VARCHAR(32) COMMENT "用户",  
>     email VARCHAR(32) COMMENT "文章类型"  
> )  
> DUPLICATE KEY(uid,username)  
> DISTRIBUTED BY HASH(uid) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.user ON duplicate\_table\_02  
> COLUMNS(uid,username,email)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "doris\_user\_test1",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
>     );  
>       
>    
>  
> 
> \--user 表  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table\_03  
> (  
>        
>     tid VARCHAR(32) COMMENT "tid id",  
>     typename VARCHAR(32) COMMENT "类型名称"  
>       
> )  
> DUPLICATE KEY(tid,typename)  
> DISTRIBUTED BY HASH(tid) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.type ON duplicate\_table\_03  
> COLUMNS(tid,typename)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "doris\_type\_test1",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
> );
> 
> \--type2 表  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table\_04  
> (  
>        
>     t\_id VARCHAR(32) COMMENT "tid id",  
>     typename VARCHAR(32) COMMENT "类型名称"  
>       
> )  
> DUPLICATE KEY(t\_id,typename)  
> DISTRIBUTED BY HASH(t\_id) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.type2 ON duplicate\_table\_04  
> COLUMNS(t\_id,typename)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "doris\_type2\_test1",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
> );
> 
> \--type3 表  
> CREATE TABLE IF NOT EXISTS example\_db.duplicate\_table\_05  
> (  
>        
>     tt\_id VARCHAR(32) COMMENT "tid id",  
>     typename VARCHAR(32) COMMENT "类型名称"  
>       
> )  
> DUPLICATE KEY(tt\_id,typename)  
> DISTRIBUTED BY HASH(tt\_id) BUCKETS 5  
> PROPERTIES("replication\_num" = "1");
> 
>   
> CREATE ROUTINE LOAD example\_db.type3 ON duplicate\_table\_05  
> COLUMNS(tt\_id,typename)  
> PROPERTIES  
> (  
> "desired\_concurrent\_number"="3",  
> "max\_batch\_interval" = "20",  
> "max\_batch\_rows" = "300000",  
> "max\_batch\_size" = "209715200",  
> "strict\_mode" = "false",  
> "format" = "json"  
> )  
> FROM KAFKA  
>     (  
>     "kafka\_broker\_list" = "192.168.12.188:9092",  
>     "kafka\_topic" = "doris\_type3\_test1",  
>     "kafka\_partitions" = "0",  
>     "kafka\_offsets" = "0"  
> );  
>       
> \--表结构变更  
>         --案例：针对testing\_table增加列
> 
> ALTER TABLE testing\_table ADD COLUMN uv BIGINT SUM DEFAULT '0' after pv;    
> 
>      
> SHOW ALTER TABLE COLUMN; ---查看进度：
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]
> 
>     ---rollup 物化索引  
> ALTER TABLE testing\_table ADD ROLLUP rollup\_id\_name(siteid, username);
> 
> SHOW ALTER TABLE ROLLUP; ---查看rollup
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]
> 
>   
> Rollup 建立完成之后可以使用 DESC table\_name ALL 查看表的 Rollup 信息。
> 
> DESC testing\_table ALL;
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]
> 
>  
> 
> 可以使用以下命令取消当前正在执行的作业:  
> CANCEL ALTER TABLE ROLLUP FROM testing\_table;
> 
> \------两个表join 测试 ：  
> \---join 分为两种：Broadcast/Shuffle Join
> 
> SELECT duplicate\_table\_03.\*,duplicate\_table\_04.\* FROM duplicate\_table\_03 JOIN duplicate\_table\_04 WHERE duplicate\_table\_03.tid = duplicate\_table\_04.t\_id;
> 
> \---使用\[broadcast\]  
> SELECT duplicate\_table\_03.\*,duplicate\_table\_04.\* FROM duplicate\_table\_03 JOIN \[broadcast\] duplicate\_table\_04 WHERE duplicate\_table\_03.tid = duplicate\_table\_04.t\_id;
> 
> \---使用 \[shuffle\] Join:  
> SELECT duplicate\_table\_03.\*,duplicate\_table\_04.\* FROM duplicate\_table\_03 JOIN \[broadcast\] duplicate\_table\_04 WHERE duplicate\_table\_03.tid = duplicate\_table\_04.t\_id;
> 
> \---使用left join  基本语法都支持  
> SELECT duplicate\_table\_03.\*,duplicate\_table\_04.\* FROM duplicate\_table\_03 left JOIN \[shuffle\] duplicate\_table\_04 on duplicate\_table\_03.tid = duplicate\_table\_04.t\_id;
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]
> 
> \---多表join
> 
> SELECT duplicate\_table\_03.\*,duplicate\_table\_04.\*,duplicate\_table\_05.\*   
> FROM duplicate\_table\_03   
> JOIN duplicate\_table\_04   
> ON duplicate\_table\_03.tid=duplicate\_table\_04.t\_id    
> JOIN duplicate\_table\_05   
> ON duplicate\_table\_04.t\_id=duplicate\_table\_05.tt\_id

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221124/13e237468f404b0eae6936909981be2c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221124/0b57ac8991074cb98af814e7e78d96ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221124/6601b4e330164aaf91cfb3dcf033e912.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221124/399586373674443992258d302fd615e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221124/39f618b6c3bd4d05b2fba6e13b79cdab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221124/46d4982e299b45688f7dd1f0060dda9f.png