Realtime Compute for Apache Flink:Object Storage Service

Batasan

Umum:

• Hanya Ververica Runtime (VVR) 11 dan versi setelahnya yang mendukung pembacaan file terkompresi (GZIP, BZIP2, XZ, DEFLATE) dari OSS. VVR 8 tidak dapat memproses file terkompresi dengan benar.

• Versi VVR sebelum 8.0.6 hanya mendukung bucket OSS dalam akun yang sama. Untuk mengakses bucket lintas akun, gunakan VVR 8.0.6 atau versi setelahnya dan konfigurasikan otentikasi bucket. Untuk detailnya, lihat Konfigurasikan otentikasi bucket.

• Pembacaan inkremental partisi baru tidak didukung.

• Akses OSS lintas wilayah tidak didukung pada VVR 11.8 dan versi sebelumnya. Bucket OSS harus berada di wilayah yang sama dengan ruang kerja Flink. OSS connector dibangun di atas antarmuka Filesystem, yang hanya mendukung satu endpoint global. Mengakses bucket OSS di wilayah berbeda menyebabkan error ketidakcocokan endpoint.

• Mulai dari VVR 11.9, Anda dapat mengakses bucket OSS yang berada di wilayah berbeda dari ruang kerja Flink Anda dengan mengonfigurasi parameter region-id dan endpoint tingkat bucket. Contoh berikut menunjukkan cara mengonfigurasi parameter tersebut:

  fs.oss.bucket.<bucket-name>.region-id: cn-shanghai
  fs.oss.bucket.<bucket-name>.endpoint: oss-cn-shanghai.aliyuncs.com

Hanya untuk tabel sink:

Format baris — Avro, CSV, JSON, dan raw — tidak didukung saat menulis ke OSS. Lihat FLINK-30635 untuk detailnya.

Sintaksis

CREATE TABLE OssTable (
  column_name1 INT,
  column_name2 STRING,
  ...
  datetime STRING,
  `hour` STRING
) PARTITIONED BY (datetime, `hour`) WITH (
  'connector' = 'filesystem',           -- wajib: harus 'filesystem'
  'path' = 'oss://<bucket>/path',       -- wajib: URI dari path OSS
  'format' = '...',                     -- wajib: orc, parquet, avro, csv, json, atau raw
  'partition.default-name' = '...',     -- opsional: nama partisi saat field partisi NULL atau kosong
  'source.monitor-interval' = '...',    -- opsional (hanya sumber): interval pemindaian file baru
  'auto-compaction' = '...'            -- opsional (hanya sink): aktifkan kompaksi otomatis setelah setiap checkpoint
);

Kolom metadata

Tabel sumber mendukung kolom metadata yang menampilkan informasi tingkat file untuk setiap baris. Definisikan kolom metadata dalam DDL Anda dengan menambahkan METADATA setelah tipe data:

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

Kolom metadata berikut tersedia:

Parameter WITH

Konfigurasikan otentikasi bucket

Hanya VVR 8.0.6 dan versi setelahnya yang mendukung otentikasi bucket.

Setelah mengatur parameter path, konfigurasikan otentikasi bucket agar Flink dapat membaca dari dan menulis ke path OSS yang ditentukan. Tambahkan konfigurasi berikut ke bagian Additional Configurations pada tab Parameters halaman Deployment Details di Konsol pengembangan Realtime Compute:

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

Ganti <bucketName> dengan nama bucket yang digunakan dalam parameter path.

Menulis ke OSS-HDFS

Tambahkan konfigurasi berikut ke bagian Other Configuration pada tab Parameters halaman Deployment Details di Konsol pengembangan Realtime Compute:

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

Konfigurasikan endpoint OSS-HDFS menggunakan salah satu metode berikut:

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

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

# Path OSS
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

Menulis ke Hadoop Distributed File System (HDFS) eksternal

Untuk path yang menggunakan skema hdfs://, tambahkan konfigurasi berikut untuk menentukan atau mengganti username akses:

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

Contoh

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 dalam milidetik
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND    -- Watermark pada kolom TIMESTAMP_LTZ
) 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

Contoh berikut menggunakan StreamingFileSink dengan OnCheckpointRollingPolicy untuk menulis ke OSS. File bagian di-rolling pada setiap 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);

Untuk menulis ke OSS-HDFS, konfigurasikan juga parameter OSS-HDFS di Additional Configurations. Lihat Menulis ke OSS-HDFS.

Langkah selanjutnya

• Connector yang didukung

• Connector Tablestore (OTS)

• Paimon untuk danau data terpadu streaming