如何在Spark Streaming SQL中使用INSERT INTO语句_开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

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'}

Insert Into