如何在Spark Streaming SQL中使用INSERT INTO語句 - E-MapReduce

StarRocks中INSERT INTO語句的使用方式和MySQL等資料庫中INSERT INTO語句的使用方式類似，但在StarRocks中，所有的資料寫入都是一個獨立的匯入作業，所以StarRocks中將INSERT INTO作為一種匯入方式介紹。本文為您介紹Insert Into匯入的使用情境、相關配置以及匯入樣本。

適用情境

INSERT INTO匯入會同步返回匯入流程的運行結果。
如果僅匯入幾條測試資料，驗證一下StarRocks系統的功能，則可以使用INSERT INTO VALUES語句。
如果將已經在StarRocks表中的資料進行ETL轉換並匯入到一個新的StarRocks表中，則可以使用INSERT INTO SELECT語句。
您可以建立一種外部表格。例如，MySQL外部表格映射一張MySQL系統中的表。然後通過INSERT INTO SELECT語句將外部表格中的資料匯入到StarRocks表中儲存。

注意事項

當執行INSERT INTO語句時，對於不符合目標表格式的資料（例如，字串超長），預設的行為是過濾。對於資料不能被過濾的業務情境，可以通過設定會話變數enable_insert_strict為true來確保當有資料被過濾的時候，INSERT INTO語句不會成功執行。
因為StarRocks的INSERT INTO複用匯入資料的邏輯，所以每一次INSERT INTO語句都會產生一個新的資料版本。因為頻繁小大量匯入操作會產生過多的資料版本，而過多的小版本會影響查詢的效能，所以不建議頻繁的使用INSERT INTO語句匯入資料或作為生產環境的日常例行匯入任務。如果有流式匯入或者小大量匯入任務的需求，則可以使用Stream Load或者Routine Load的方式進行匯入。

基本操作

文法

INSERT INTO table_name
[ PARTITION (p1, ...) ]
[ WITH LABEL label]
[ (column [, ...]) ]
[ [ hint [, ...] ] ]
{ VALUES ( { expression | DEFAULT } [, ...] ) [, ...] | query }

參數描述如下表所示。

參數	描述
table_name	匯入資料的目標表的名稱。填寫形式為db_name.table_name。
partitions	指定待匯入的分區，必須是table_name表中存在的分區，多個分區名稱用逗號（,）分隔。如果指定目標資料分割，則只會匯入符合目標資料分割的資料。如果沒有指定，則預設值為table_name表中所有的分區。
label	為insert作業指定一個Label，Label是該INSERT INTO匯入作業的標識。每個匯入作業，都有一個在單DataBase內部唯一的Label。重要建議指定Label，而不是由系統自動分配。如果由系統自動分配，在INSERT INTO語句執行過程中，因網路錯誤導致串連斷開等，則無法得知INSERT INTO是否成功。如果指定Label，則可以再次通過Label查看任務結果。
column_name	指定的目的列，必須是table_name表中存在的列。匯入表的目標列，可以以任意的順序存在。如果沒有指定目標列，則預設值是table_name表的所有列。如果匯入表中的某個列不在目標列中，則這個列需要有預設值，否則INSERT INTO會失敗。如果查詢語句的結果列類型與目標列的類型不一致，則會調用隱式類型轉化，如果不能進行轉化，則INSERT INTO語句會報文法解析錯誤。
expression	需要賦值給某個列的對應運算式。
default	讓對應列使用預設值。
query	一個普通查詢，查詢的結果會寫入到目標中。查詢語句支援任意StarRocks支援的SQL查詢語句。
values	通過VALUES語句插入一條或者多條資料。重要 VALUES方式僅適用於匯入幾條資料作為Demo的情況，完全不適用於任何測試和生產環境。StarRocks系統本身也不適合單條資料匯入的情境。建議使用INSERT INTO SELECT的方式進行大量匯入。

匯入結果介紹

INSERT INTO本身就是一個SQL命令，其返回結果如下所示：

執行成功

樣本1
執行insert into tbl1 select * from empty_tbl;匯入語句。返回結果如下。
```
Query OK, 0 rows affected (0.02 sec)
```

樣本2

執行insert into tbl1 select * from tbl2;匯入語句。返回結果如下。

Query OK, 4 rows affected (0.38 sec)
{'label':'insert_8510c568-9eda-4173-9e36-6adc7d35291c', 'status':'visible', 'txnId':'4005'}

樣本3
執行insert into tbl1 with label my_label1 select * from tbl2;匯入語句。返回結果如下。
```
Query OK, 4 rows affected (0.38 sec)
{'label':'my_label1', 'status':'visible', 'txnId':'4005'}
```

樣本4

執行insert into tbl1 select * from tbl2;匯入語句。返回結果如下。

Query OK, 2 rows affected, 2 warnings (0.31 sec)
{'label':'insert_f0747f0e-7a35-46e2-affa-13a235f4020d', 'status':'visible', 'txnId':'4005'}

或者返回如下結果。

Query OK, 2 rows affected, 2 warnings (0.31 sec)
{'label':'insert_f0747f0e-7a35-46e2-affa-13a235f4020d', 'status':'committed', 'txnId':'4005'}

返回結果中涉及的參數描述如下表所示。

參數	描述
rows affected	表示總共有多少行資料已匯入。
warnings	表示被過濾的行數。
status	匯入資料是否可見。參數值如下： visible：表示可見。 committed：表示不可見。
txnId	insert對應的匯入事務的ID。
err	顯示一些非預期錯誤。當需要查看被過濾的行時，您可以使用`SHOW LOAD`語句，返回結果中的URL可以用於查詢錯誤的資料。

執行失敗

執行失敗表示沒有成功匯入任何資料。

例如，執行insert into tbl1 select * from tbl2 where k1 = "a";匯入語句。返回結果如下。

ERROR 1064 (HY000): all partitions have no load data. url: [http://10.74.167.16:8042/api/_load_error_log?file=__shard_2/error_log_insert_stmt_ba8bb9e158e4879-ae8de8507c0bf8a2_ba8bb9e158e4879_ae8de8507c0bf8a2](http://10.74.**.**:8042/api/_load_error_log?file=__shard_2/error_log_insert_stmt_ba8bb9e158e4879-ae8de8507c0bf8a2_ba8bb9e158e4879_ae8de8507c0bf8a2)

其中，返回資訊中的ERROR 1064 (HY000): all partitions have no load data顯示失敗原因，後面的URL可以用於查詢錯誤的資料。

匯入樣本

建立資料庫與資料表

執行以下命令，建立資料庫。

CREATE DATABASE IF NOT EXISTS load_test;

執行以下命令，指定資料庫。
```
USE load_test;
```

執行以下命令，建立資料表。

CREATE TABLE insert_wiki_edit
(
    event_time DATETIME,
    channel VARCHAR(32) DEFAULT '',
    user VARCHAR(128) DEFAULT '',
    is_anonymous TINYINT DEFAULT '0',
    is_minor TINYINT DEFAULT '0',
    is_new TINYINT DEFAULT '0',
    is_robot TINYINT DEFAULT '0',
    is_unpatrolled TINYINT DEFAULT '0',
    delta INT SUM DEFAULT '0',
    added INT SUM DEFAULT '0',
    deleted INT SUM DEFAULT '0'
)
AGGREGATE KEY(event_time, channel, user, is_anonymous, is_minor, is_new, is_robot, is_unpatrolled)
PARTITION BY RANGE(event_time)
(
    PARTITION p06 VALUES LESS THAN ('2015-09-12 06:00:00'),
    PARTITION p12 VALUES LESS THAN ('2015-09-12 12:00:00'),
    PARTITION p18 VALUES LESS THAN ('2015-09-12 18:00:00'),
    PARTITION p24 VALUES LESS THAN ('2015-09-13 00:00:00')
)
DISTRIBUTED BY HASH(user) BUCKETS 10
PROPERTIES("replication_num" = "1");

通過VALUES匯入資料

執行以下命令，通過VALUES語句匯入資料。

INSERT INTO insert_wiki_edit VALUES("2015-09-12 00:00:00","#en.wikipedia","GELongstreet",0,0,0,0,0,36,36,0),("2015-09-12 00:00:00","#ca.wikipedia","PereBot",0,1,0,1,0,17,17,0);

返回資訊如下。

Query OK, 2 rows affected (0.29 sec)
{'label':'insert_1f12c916-5ff8-4ba9-8452-6fc37fac2e75', 'status':'visible', 'txnId':'601'}

通過SELECT匯入資料

執行以下命令，通過SELECT語句匯入資料。

INSERT INTO insert_wiki_edit WITH LABEL insert_load_wikipedia SELECT * FROM routine_wiki_edit;

返回資訊如下。

Query OK, 18203 rows affected (0.40 sec)
{'label':'insert_load_wikipedia', 'status':'visible', 'txnId':'618'}

E-MapReduce：Insert Into