--- title: Databend 在 minio 环境使用 copy 命令 | 新手篇（3） description: Databend 在 minio 环境使用 copy 命令 | 新手篇（3） slug: category-engineering/use-copy-into-databend date: 2022-12-29 tags: [新手篇] category: Engineering cover_url: beginner-03.png image: beginner-03.png authors: - wubx --- Copy 命令是 Databend 批量导入 | 卸载数据的灵魂工具。目前 copy 可以做到单机内按 CPU core 数并发，马上就支持按集群内并发。Copy 可以结合 bucket 和 内置的 stage 使用，这块功能比较丰富，接下来给大家具体讲解一下。 {/* truncate */} Databend 批量数据装载和卸载都是通过 copy 命令来实现。 ## Copy 命令讲解 ### Copy 命令格式 功能：将数据导入到 table 中，注意导入的前提是先将 table 创建好。 ```bash COPY INTO [.] FROM { internalStage | externalStage | externalLocation } [ FILES = ( '' [ , '' ] [ , ... ] ) ] [ 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 使用 1. 创建 内部 stage ```Bash MySQL [default]> create stage my_stage; Query OK, 0 rows affected (0.030 sec) ``` 2. 查看有哪些 stage ```Bash 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 下面 ```Bash $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/stage/presign presign upload @my_stage/t1.jsoncurl -X PUT -T t1.csv '' ``` 其中 curl 把文件 put 到 stage 只有两个参数：一个用 -H 约定 stage_name 另一个用 -F 约定 upload 的文件名，后面对应的 URL 前 @ 部分注意加上用户名和密码，这里用的 root 在 localhost 下面空密码。 4. 查看 my_stage 的内容 ```Bash 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) ``` 5. 使用 copy 从 my_stage 导入数据 ```Bash 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 后面容易把 = 号给忘了。 6. 内部 stage 文件删除 ```Bash MySQL [default]> remove @my_stage/t1.json; Query OK, 0 rows affected (0.028 sec) ``` 7. 删除 stage ```Bash 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： ```Bash 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 这个参数目前发现国内的阿里云，火山引擎需要，其它的不需要。 2. Stage 查看 ```Bash 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) ``` 3. 外部 stage 上传文件 对于 bucket 上传文件可以支持直接通过 bucket 写入，如果给的权限够大，也可以使用 curl 的方式 put stage，如： ```Bash 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"]} ``` 也可以使用： ```Bash #minio 的别名操作 mc copy t1.json mymino/wubx/ ``` 接下来的使用的内部 stage 的使用方式是一样的。 4. 通过 copy 从 ex_wubx 这个 stage 中导入数据 ```Bash 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) ``` 5. 删除 删除文件和 外部 stage 同内部 stage 一样。 ## 总结 Copy 命令语法比较复杂，这里只是讲解两个 stage 的使用，对于创建 stage 时也可以把 FILE_FORMAT 带上，这样后面写 copy into 可以省略对应的 FILE_FORMAT。例如： ```Bash 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。 - [Databend Website](https://docs.databend.cn) - [GitHub Discussions](https://github.com/databendlabs/databend/discussions) - [Twitter](https://twitter.com/Datafuse_Labs) - [Slack Channel](https://link.databend.com/join-slack)