对象存储OBS结果表

功能描述

FileSystem sink用于将数据输出到分布式文件系统HDFS或者对象存储服务OBS等文件系统。适用于数据转储、大数据分析、备份或活跃归档、深度或冷归档等场景。

考虑到输入流可以是无界的，每个桶中的数据被组织成有限大小的Part文件。完全可以配置为基于时间的方式往桶中写入数据，比如可以设置每个小时的数据写入一个新桶中。即桶中将包含一个小时间隔内接收到的记录。

桶目录中的数据被拆分成多个Part文件。对于相应的接收数据的桶的Sink的每个Subtask，每个桶将至少包含一个Part文件。将根据配置的滚动策略来创建其他Part文件。对于Row Formats默认的策略是根据Part文件大小进行滚动，需要指定文件打开状态最长时间的超时以及文件关闭后的非活动状态的超时时间。对于Bulk Formats在每次创建Checkpoint时进行滚动，并且用户也可以添加基于大小或者时间等的其他条件。更多信息参考文件系统 SQL 连接器

• 在STREAMING模式下使用FileSink需要开启Checkpoint功能。Part文件只在Checkpoint成功时生成。如果没有开启Checkpoint功能，文件将永远停留在in-progress或者pending的状态，并且下游系统将不能安全读取该文件数据。
• sink end算子的接受记录数为checkpoint的个数，非实际的发送数据，实际发送数据量请参考streaming-writer或StreamingFileWriter算子的记录数。

注意事项

请在Flink“作业编辑”页面选择“运行参数配置”，选择“开启Checkpoint”，否则会导致FileSystem结果表无法写入数据。

语法格式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE TABLE sink_table (
   name string,
   num INT,
   p_day string,
   p_hour string
) partitioned by (p_day, p_hour) WITH (
   'connector' = 'filesystem',
   'path' = 'obs://*** ',
   'format' = 'parquet',
   'auto-compaction' = 'true'
);

使用说明

• 滚动策略

  RollingPolicy 定义了何时关闭给定的In-progress Part文件，并将其转换为Pending状态，然后再转换为Finished状态。 Finished状态的文件，可供查看并且可以保证数据的有效性，在出现故障时不会恢复。

  在 STREAMING模式下，滚动策略结合Checkpoint间隔（到下一个Checkpoint成功时，文件的Pending状态才转换为 Finished 状态），共同控制Part文件对下游readers是否可见以及这些文件的大小和数量。详见滚动策略相关参数说明。

• Part文件生命周期

  为了在下游使用 FileSink 作为输出，需要了解生成的输出文件的命名和生命周期。

  Part 文件可以处于以下三种状态中的任意一种：

  • In-progress：当前正在写入的 Part 文件处于 in-progress 状态
  • Pending：由于指定的滚动策略）关闭 in-progress 状态文件，并且等待提交
  • Finished：流模式(STREAMING)下的成功的 Checkpoint 或者批模式(BATCH)下输入结束，文件的Pending状态转换为 Finished 状态

  只有 Finished 状态下的文件才能被下游安全读取，并且保证不会被修改。

  默认的，Part文件命名策略如下:

  • In-progress / Pending：part-<uid>-<partFileIndex>.inprogress.uid
  • Finished：part-<uid>-<partFileIndex>

  当Sink Subtask实例化时，uid是一个分配给 Subtask 的随机ID值。uid不具有容错机制，所以当Subtask从故障恢复时，uid会重新生成。

• 文件合并

  FileSink 开始支持已经提交Pending文件的合并，从而允许应用设置一个较小的时间周期并且避免生成大量的小文件。

  这一功能开启后，在文件转为Pending状态与文件最终提交之间会进行文件合并。这些Pending状态的文件将首先被提交为一个以.开头的临时文件。这些临时文件随后将会按照用户指定的策略和合并方式进行合并，最终生成合并后的Pending状态的文件。 然后这些文件将被发送给Committer并提交为正式文件，在这之后，原始的临时文件也会被删除掉。

• 分区功能

  Filesystem sink支持分区功能，通过partitioned by语法根据选择的字段进行分区。示例如下：

  path
  └── datetime=2022-06-25
      └── hour=10
          ├── part-0.parquet
          ├── part-1.parquet
  └── datetime=2022-06-26
      └── hour=16
          ├── part-0.parquet
      └── hour=17
          ├── part-0.parquet

  分区和文件一样，也需要进行提交，通知下游应用可以安全地读取分区内的文件。Filesystem sink提供多种提交配置策略。

参数说明

示例1

create table orders(
  name string,
  num INT
) with (
  'connector' = 'datagen',
  'rows-per-second' = '100', 
  'fields.name.kind' = 'random', 
  'fields.name.length' = '5' 
);

CREATE TABLE sink_table (
   name string,
   num INT
) WITH (
   'connector' = 'filesystem',
   'path' = 'obs://bucketName/fileName',
   'format' = 'csv',
   'sink.rolling-policy.file-size'='128m',
   'sink.rolling-policy.rollover-interval'='30 min'
);
INSERT into sink_table SELECT * from orders;

示例2

create table orders(
  name string,
  num INT
) with (
  'connector' = 'datagen',
  'rows-per-second' = '100', 
  'fields.name.kind' = 'random', 
  'fields.name.length' = '5' 
);

CREATE TABLE sink_table (
   name string,
   num INT
) WITH (
   'connector' = 'filesystem',
   'path' = 'obs://bucketName/fileName',
   'format' = 'parquet',
   'sink.rolling-policy.file-size'='128m',
   'sink.rolling-policy.rollover-interval'='30 min',
   'auto-compaction'='true',
   'compaction.file-size'='100m'

);
INSERT into sink_table SELECT * from orders;