Realtime Compute for Apache Flink:Object Storage Service

Limitations

General:

• Only Ververica Runtime (VVR) 11 and later support reading compressed files (GZIP, BZIP2, XZ, DEFLATE) from OSS. VVR 8 cannot process compressed files correctly.

• VVR versions earlier than 8.0.6 only support OSS buckets in the same account. To access buckets across accounts, use VVR 8.0.6 or later and configure bucket authentication. For details, see Configure bucket authentication.

• Incremental reading of new partitions is not supported.

Sink tables only:

Row formats — Avro, CSV, JSON, and raw — are not supported when writing to OSS. See FLINK-30635 for details.

Syntax

CREATE TABLE OssTable (
  column_name1 INT,
  column_name2 STRING,
  ...
  datetime STRING,
  `hour` STRING
) PARTITIONED BY (datetime, `hour`) WITH (
  'connector' = 'filesystem',           -- required: must be 'filesystem'
  'path' = 'oss://<bucket>/path',       -- required: URI of the OSS path
  'format' = '...',                     -- required: orc, parquet, avro, csv, json, or raw
  'partition.default-name' = '...',     -- optional: partition name when partition field is NULL or empty
  'source.monitor-interval' = '...',    -- optional (source only): interval to scan for new files
  'auto-compaction' = '...'            -- optional (sink only): enable automatic compaction after each checkpoint
);

Metadata columns

Source tables support metadata columns that expose file-level information about each row. Define a metadata column in your DDL by adding METADATA after the data type:

CREATE TABLE MyUserTableWithFilepath (
  column_name1 INT,
  column_name2 STRING,
  `file.path` STRING NOT NULL METADATA
) WITH (
  'connector' = 'filesystem',
  'path' = 'oss://<bucket>/path',
  'format' = 'json'
)

The following metadata columns are available:

WITH parameters

Configure bucket authentication

Only VVR 8.0.6 and later support bucket authentication.

After setting the path parameter, configure bucket authentication so Flink can read from and write to the specified OSS path. Add the following to the Additional Configurations section on the Parameters tab of the Deployment Details page in the Realtime Compute development console:

fs.oss.bucket.<bucketName>.accessKeyId: <your-access-key-id>
fs.oss.bucket.<bucketName>.accessKeySecret: <your-access-key-secret>

Replace <bucketName> with the bucket name used in the path parameter.

Write to OSS-HDFS

Add the following configuration to the Additional Configurations section on the Parameters tab of the Deployment Details page in the Realtime Compute development console:

fs.oss.jindo.buckets: <bucket-names>
fs.oss.jindo.accessKeyId: <your-access-key-id>
fs.oss.jindo.accessKeySecret: <your-access-key-secret>

Configure the OSS-HDFS endpoint using one of the following methods:

fs.oss.jindo.endpoint: <oss-hdfs-endpoint>

oss://<bucket-name>.<oss-hdfs-endpoint>/<directory>

# OSS path
oss://jindo-test.cn-beijing.oss-dls.aliyuncs.com/<directory>

# Additional Configurations
fs.oss.jindo.buckets: jindo-test,jindo-test.cn-beijing.oss-dls.aliyuncs.com

Writing to an external Hadoop Distributed File System (HDFS)

For paths using the hdfs:// scheme, add the following to specify or switch the access username:

containerized.taskmanager.env.HADOOP_USER_NAME: hdfs
containerized.master.env.HADOOP_USER_NAME: hdfs

Examples

CREATE TEMPORARY TABLE fs_table_source (
  `id` INT,
  `name` VARCHAR
) WITH (
  'connector' = 'filesystem',
  'path' = 'oss://<bucket>/path',
  'format' = 'parquet'
);

CREATE TEMPORARY TABLE blackhole_sink (
  `id` INT,
  `name` VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink SELECT * FROM fs_table_source;

CREATE TABLE datagen_source (
  user_id STRING,
  order_amount DOUBLE,
  ts BIGINT,                                               -- Timestamp in milliseconds
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND    -- Watermark on TIMESTAMP_LTZ column
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE fs_table_sink (
  user_id STRING,
  order_amount DOUBLE,
  dt STRING,
  `hour` STRING
) PARTITIONED BY (dt, `hour`) WITH (
  'connector' = 'filesystem',
  'path' = 'oss://<bucket>/path',
  'format' = 'parquet',
  'partition.time-extractor.timestamp-pattern' = '$dt $hour:00:00',
  'sink.partition-commit.delay' = '1 h',
  'sink.partition-commit.trigger' = 'partition-time',
  'sink.partition-commit.watermark-time-zone' = 'Asia/Shanghai',
  'sink.partition-commit.policy.kind' = 'success-file'
);

INSERT INTO fs_table_sink
SELECT
  user_id,
  order_amount,
  DATE_FORMAT(ts_ltz, 'yyyy-MM-dd'),
  DATE_FORMAT(ts_ltz, 'HH')
FROM datagen_source;

CREATE TABLE datagen_source (
  user_id STRING,
  order_amount DOUBLE
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE fs_table_sink (
  user_id STRING,
  order_amount DOUBLE
) WITH (
  'connector' = 'filesystem',
  'path' = 'oss://<bucket>/path',
  'format' = 'parquet'
);

INSERT INTO fs_table_sink SELECT * FROM datagen_source;

DataStream API

The following example uses StreamingFileSink with OnCheckpointRollingPolicy to write to OSS. Part files are rolled on every checkpoint.

String outputPath = "oss://<bucket>/path";

final StreamingFileSink<Row> sink =
    StreamingFileSink.forRowFormat(
            new Path(outputPath),
            (Encoder<Row>) (element, stream) -> {
                out.println(element.toString());
            })
        .withRollingPolicy(OnCheckpointRollingPolicy.build())
        .build();

outputStream.addSink(sink);

To write to OSS-HDFS, also configure the OSS-HDFS parameters in Additional Configurations. See Write to OSS-HDFS.

What's next

• Supported connectors

• Tablestore (OTS) connector

• Paimon for streaming data lakehouses