博客

Databend 在 minio 环境使用 copy 命令 | 新手篇(3)

wubx12月 29, 2022

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
CSVFILE_FORMAT = (type='CSV' field_delimiter=',' skip_header=1 compression=auto)
TSVFILE_FORMAT = (type='TSV' field_delimiter=',' skip_header=1 compression=auto)
ndjsonFILE_FORMAT = (type='NDJSON' compression=auto)
parquetFILE_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 使用

  1. 创建 内部 stage

    MySQL [default]> create stage my_stage;
    Query OK, 0 rows affected (0.030 sec)
  2. 查看有哪些 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)
  3. 利用 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 下面空密码。

  1. 查看 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)
  1. 使用 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 后面容易把 = 号给忘了。

  1. 内部 stage 文件删除
MySQL [default]> remove @my_stage/t1.json;
Query OK, 0 rows affected (0.028 sec)
  1. 删除 stage
MySQL [default]> drop stage my_stage;
Query OK, 0 rows affected (0.036 sec)

外部 stage 使用

  1. 外部 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 这个参数目前发现国内的阿里云,火山引擎需要,其它的不需要。

  1. 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)
  1. 外部 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 的使用方式是一样的。

  1. 通过 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)
  1. 删除

删除文件和 外部 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。

分享本篇文章

订阅我们的新闻简报

及时了解功能发布、产品规划、支持服务和云服务的最新信息!