Filesystem 连接器提供对常见的文件系统（如 CloudFS）的读写能力，支持做数据源表和结果表。您可以创建 source 流从文件系统目录下获取数据，作为作业的输入数据；也可以将作业输出数据写入到文件系统指定目录下。

DDL 定义 #

用作数据源（Source） #

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

用作数据目的（Sink） #

CREATE TABLE filesystem_sink (
    name String,
    score INT
 ) WITH (
     'connector' = 'filesystem',
     'path' = '<yourfilepath>',
     '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;