Copy 命令是 Databend 批量导入 | 卸载数据的灵魂工具。目前 copy 可以做到单机内按 CPU core 数并发，马上就支持按集群内并发。Copy 可以结合 bucket 和 内置的 stage 使用，这块功能比较丰富，接下来给大家具体讲解一下。
Databend 批量数据装载和卸载都是通过 copy 命令来实现。
Copy 命令讲解
Copy 命令格式
功能：将数据导入到 table 中，注意导入的前提是先将 table 创建好。
COPY INTO [<database>.]<table_name>
FROM { internalStage | externalStage | externalLocation }
[ FILES = ( '<file_name>' [ , '<file_name>' ] [ , ... ] ) ]
[ PATTERN = '<regex_pattern>' ]
[ FILE_FORMAT = ( TYPE = { CSV | JSON | NDJSON | PARQUET }
[ formatTypeOptions ] ) ]
[ copyOptions ]
命令核心：
From 指定来源：内部 stage, 外部 stage，也可以直接访问外面的 bucket。
FILES：可选项，一个文件名的列表，这个有一个约束，列表里的文件需要是同一类文件
PATTERN: 支持一个正则的文件名表达
FILE_FORMAT: 文件支持 csv, tsv, ndjson, parquet , 其中对于 csv，tsv 需要定义 formatTypeOption 对应如下：
| formatTypeOptions | |
|---|---|
| CSV | FILE_FORMAT = (type='CSV' field_delimiter=',' skip_header=1 compression=auto) |
| TSV | FILE_FORMAT = (type='TSV' field_delimiter=',' skip_header=1 compression=auto) |
| ndjson | FILE_FORMAT = (type='NDJSON' compression=auto) |
| parquet | FILE_FORMAT = (type='parquet') |
field_delimiter: 指定字段间隔符，csv 默认是 ',' , tsv 默认是'\t'
record_delimiter: 指定换行符, 建议不指定，copy 命令会自动识别。
skip_header: 指定跳过头部的几行
compresssion: auto 指定自动识别压缩。可以使用的值：GZIP | BZ2 | BROTLI | ZSTD | DEFALTE | RAW_DEFLATE
escape: 对于一些非标准的 csv 或是文件中有 json 字段，导入失败时建议在 File_format 中添加：escape='\' 尝试
copyOptions: 可选项。有三个参数：
| 参数 | 用处 |
|---|---|
| size_limit | 限定只导入多少行，一般用于测试阶段。默认是 0 表示不限制 |
| purge | 布尔型，如果为：true 会把 copy 成功的文件删除。默认是：False |
| force | 布尔型。Databend 默认 Copy into 对于导入成功的文件不会再次导入，如果需要重复导入，这个参数需要设置成 force 为 true，这个参数不要和 purge 一起使用。一般也是测蔗环境使用。 |
例如 copy 后需要把文件删除：purge=true
接下来我们重点实战一下：基于 内部 stage 和 外部 stage 的 copy into。
Stage 讲解
什么 是 stage?
在 Databend 中 Stage 是用来暂存数据的一个空间。通常是对象存储中一个 bucket 或是 bucket 下面的某个目录。从这个 bucket 的产生情况分为：
外部 stage : 用户 Account 下创建的 bucket ， 非 Databend-query 连接的 bucket。 使用外部 bucket 需要知道 bucket 路径，对应的 endpoint_url ， id, key 等信息。
内部 stage: 是指 Databend-query 连接的 bucket 下，用户通过 create stage <stage_name> 直接创建的目录，该目录 Databend-query 拥有读取权限。
内部 stage 使用
创建 内部 stage
MySQL [default]> create stage my_stage;
Query OK, 0 rows affected (0.030 sec)查看有哪些 stage
MySQL [default]> show stages;
+----------+------------+-----------------+--------------------+---------+
| name | stage_type | number_of_files | creator | comment |
+----------+------------+-----------------+--------------------+---------+
| my_stage | Internal | 0 | 'root'@'127.0.0.1' | |
+----------+------------+-----------------+--------------------+---------+
1 row in set (0.027 sec)利用 curl 向 stage 上传文件
把 t1.json 上传到 my_stage 下面
$cat t1.json
{"id": 1011,"name": "geek","passion": "codingx"}
{"id": 1012,"name": "geek","passion": "codingx"}
# 私有部署环境支持
$curl -H "stage_name:my_stage" -F "upload=@t1.json" -XPUT "http://root:@localhost:8000/v1/upload_to_stage"
{"id":"2ca946ec-4400-4190-a00d-335a1eee8337","stage_name":"my_stage","state":"SUCCESS","files":["t1.json"]}
# 云环境和公有环境都可以使用
# 参考：https://docs.databend.cn/sql/sql-commands/ddl/presign/presign
presign upload @my_stage/t1.jsoncurl -X PUT -T t1.csv '<presign-url>'
其中 curl 把文件 put 到 stage 只有两个参数：一个用 -H 约定 stage_name 另一个用 -F 约定 upload 的文件名，后面对应的 URL 前 @ 部分注意加上用户名和密码，这里用的 root 在 localhost 下面空密码。
- 查看 my_stage 的内容
MySQL [(none)]> list @my_stage;
+---------+------+----------------------------------+-------------------------------+---------+
| name | size | md5 | last_modified | creator |
+---------+------+----------------------------------+-------------------------------+---------+
| t1.json | 98 | 6cdc83ea7b8d033bf74d026a666a99f2 | 2022-10-20 15:07:48.000 +0000 | NULL |
+---------+------+----------------------------------+-------------------------------+---------+
1 row in set (0.032 sec)
- 使用 copy 从 my_stage 导入数据
MySQL [default]> create table t1(id int, name varchar, passion varchar);
Query OK, 0 rows affected (0.050 sec)
MySQL [default]> copy into t1 from @my_stage file_format=(type='ndjson');
Query OK, 0 rows affected (0.113 sec)
MySQL [default]> select * from t1;
+------+------+---------+
| id | name | passion |
+------+------+---------+
| 1011 | geek | codingx |
| 1012 | geek | codingx |
+------+------+---------+
2 rows in set (0.045 sec)
这个命令易错的地方是 file_format 后面容易把 = 号给忘了。
- 内部 stage 文件删除
MySQL [default]> remove @my_stage/t1.json;
Query OK, 0 rows affected (0.028 sec)
- 删除 stage
MySQL [default]> drop stage my_stage;
Query OK, 0 rows affected (0.036 sec)
外部 stage 使用
- 外部 stage 创建
外部 stage 支持跨云，使用外部 bucket，但需要用户拥有外部 bucket 最小权限为：read, list 两个权限。下面例子：假设 Databend 部署在了 MinIO，数据源在阿里云平台的 OSS 对象存储，可以通过以下命令在 MinIO 中对 OSS 创建一个外部 stage：
create stage my_stage url='s3://myfile'
connection=( \
endpoint_url='https://***.aliyuncs.com/' \
access_key_id='***' \
secret_access_key='***'\
ENABLE_VIRTUAL_HOST_STYLE='ture);
更多参数参考：https://docs.databend.cn/sql/sql-commands/ddl/stage/ddl-create-stage#externalstageparams
ENABLE_VIRTUAL_HOST_STYLE 这个参数目前发现国内的阿里云，火山引擎需要，其它的不需要。
- Stage 查看
MySQL [default]> show stages;
+----------+------------+-----------------+--------------------+---------+
| name | stage_type | number_of_files | creator | comment |
+----------+------------+-----------------+--------------------+---------+
| ex_wubx | External | NULL | 'root'@'127.0.0.1' | |
| my_stage | Internal | 0 | 'root'@'127.0.0.1' | |
+----------+------------+-----------------+--------------------+---------+
2 rows in set (0.027 sec)
- 外部 stage 上传文件
对于 bucket 上传文件可以支持直接通过 bucket 写入，如果给的权限够大，也可以使用 curl 的方式 put stage，如：
curl -H "stage_name:ex_wubx" -F "upload=@t1.json" -XPUT "http://root:@localhost:8000/v1/upload_to_stage"
{"id":"2af385f8-c227-462a-9002-f5a72a9194ac","stage_name":"ex_wubx","state":"SUCCESS","files":["t1.json"]}
也可以使用：
#minio 的别名操作
mc copy t1.json mymino/wubx/
接下来的使用的内部 stage 的使用方式是一样的。
- 通过 copy 从 ex_wubx 这个 stage 中导入数据
MySQL [(none)]> list @ex_wubx;
+---------+------+----------------------------------+-------------------------------+---------+
| name | size | md5 | last_modified | creator |
+---------+------+----------------------------------+-------------------------------+---------+
| t1.json | 98 | 6cdc83ea7b8d033bf74d026a666a99f2 | 2022-10-20 15:32:01.000 +0000 | NULL |
+---------+------+----------------------------------+-------------------------------+---------+
1 row in set (0.036 sec)
MySQL [(none)]> select * from t1;
+------+------+---------+
| id | name | passion |
+------+------+---------+
| 1011 | geek | codingx |
| 1012 | geek | codingx |
+------+------+---------+
2 rows in set (0.045 sec)
#因为 t1.json 对表 t1 导入过一次，所以这里使用了 force=true 这个参数
MySQL [(none)]>copy into t1 from @ex_wubx file_format=(type='ndjson') force=true;
Query OK, 0 rows affected (0.138 sec)
MySQL [(none)]> select * from t1;
+------+------+---------+
| id | name | passion |
+------+------+---------+
| 1011 | geek | codingx |
| 1012 | geek | codingx |
| 1011 | geek | codingx |
| 1012 | geek | codingx |
+------+------+---------+
4 rows in set (0.048 sec)
- 删除
删除文件和 外部 stage 同内部 stage 一样。
总结
Copy 命令语法比较复杂，这里只是讲解两个 stage 的使用，对于创建 stage 时也可以把 FILE_FORMAT 带上，这样后面写 copy into 可以省略对应的 FILE_FORMAT。例如：
create stage my_stage FILE_FORMAT=(type='ndjson');
# 上传文件# 省掉 file_format 定义
copy into t1 from @my_stage force=true;
最后给大家讲一个秘密：为什么要搞一个 stage ? 真实的原因是在云上流量费非常的贵，对象存储在云同一个 Region 中没有 VPC 的概念，这样云原生数仓一般会借助于对象存储进行数据交换，而且对象存储非常便宜，借助 Stage 这个概念，可以把对象存储的操作简化。那么现在谁在利用类似 stage 的方式在中转数据呢？例如：数仓产品有 Snowflake , Databend，数据 CDC、ETL 类工具如：Fivetran, Airbyte 等等产品。用好对象存储中转数据，搞不好可以给公司节省一个出国游的费用。
Connect With Us
Databend 是一款开源、弹性、低成本，基于对象存储也可以做实时分析的新式数仓。期待您的关注，一起探索云原生数仓解决方案，打造新一代开源 Data Cloud。
