火山引擎对象存储 TOS（Torch Object Storage）是火山引擎提供的海量、安全、低成本、易用、高可靠、高可用的分布式云存储服务。通过 Flink 内置的 Filesytem Connector，您可以轻松访问和管理火山引擎 TOS 上的数据。

DDL 定义 #

用作数据源（Source） #

将 TOS 文件/目录作为数据源非常简单，只要用如下建表语句，指定 filesystem connector，以及 tos://开头的文件路径，以及文件格式即可：

CREATE TABLE filesystem_source (
    name String,
    score INT
 ) WITH (
     'connector' = 'filesystem',
     'path' = 'tos://<bucket>/<path>',
     'format' = 'json'
 );

用作数据目的（Sink） #

将 TOS 文件/目录作为数据下游，用如下建表语句，指定 filesystem connector，以及 tos://开头的文件路径、文件格式，以及文件生成规则：

CREATE TABLE filesystem_sink (
    name String,
    score INT
 ) WITH (
     'connector' = 'filesystem',
     'path' = 'tos://<bucket>/<path>',
     'sink.rolling-policy.file-size' = '1M',
     'sink.rolling-policy.rollover-interval' = '10 min',
     'format' = 'json' 
 );

WITH 参数 #

示例代码 #

• 结果表

CREATE TABLE datagen_source (
    order_id bigint,
    order_product_id bigint,
    order_customer_id bigint,
    order_status varchar
  )
WITH
  (
    'connector' = 'datagen',
    'rows-per-second' = '1',
    'fields.order_status.length' = '3',
    'fields.order_id.min' = '1',
    'fields.order_id.max' = '1000',
    'fields.order_product_id.min' = '1',
    'fields.order_product_id.max' = '100',
    'fields.order_customer_id.min' = '1',
    'fields.order_customer_id.max' = '1000'
  );

CREATE TABLE file_sink (
    order_id bigint,
    order_product_id bigint,
    order_customer_id bigint,
    order_status varchar
  )
WITH
  (
    'connector' = 'filesystem',
    'path' = 'tos://doc-test/test_write_tos/',
    'format' = 'json'
  );

INSERT INTO file_sink
SELECT * FROM datagen_source;

• 源表

CREATE TABLE file_source(
  order_id bigint, 
  order_product_id bigint,
  order_customer_id bigint,
  order_status varchar
) WITH (
    'connector' = 'filesystem',
    'path' = 'tos://doc-test/test_write_tos/',
    'format' = 'json'
);

create table print_sink (
  order_id bigint, 
  order_product_id bigint,
  order_customer_id bigint,
  order_status varchar
  )
with
  ('connector' = 'print');

insert into print_sink
SELECT * FROM file_source;