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,外部位置(使用比较麻烦,本篇暂不做讲解)
FILES :可选项, 一个文件名的列表,这个有一个约束,列表里的文件需要是同一类文件
PATTERN: 支持一个正则的文件名表达
FILE_FORMAT: 文件支持 csv, tsv, ndjson, parquet , 其中对于 csv, tsv 需要定义 formatTypeOption 对应如下:
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: 可选项。有三个参数:
例如 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)
复制代码
2 查看有哪些 stage
MySQL [default]> create stage my_stage;
Query OK, 0 rows affected (0.030 sec)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)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"]}
复制代码
其中 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://databend.rs/doc/reference/sql/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 差不多讲完了。
最后给大家讲一个秘密:为什么要搞一个 stage ? 真实的原因是在云上流量费非常的贵,对象存储在云同一个 Region 中没有 VPC 的概念,这样云原生数仓一般会借助于对象存储进行数据交换,而且对象存储非常便宜,借助 Stage 这个概念,可以把对象存储的操作简化。那么现在谁在利用类似 stage 的方式在中转数据呢?例如: 数仓产品有 Snowflake , Databend, 数据 CDC、 ETL 类工具如:Fivetran, Airbyte 等产品。用好对象存储中转数据,搞不好可以给公司节省一个出国游的费用。
关于 Databend
Databend 是一款开源、弹性、低成本,基于对象存储也可以做实时分析的新式数仓。期待您的关注,一起探索云原生数仓解决方案,打造新一代开源 Data Cloud。
Databend 文档:https://databend.rs/
Twitter:https://twitter.com/Datafuse_Labs
Slack:https://datafusecloud.slack.com/
Wechat:Databend
GitHub :https://github.com/datafuselabs/databend
评论