Realtime Compute for Apache Flink:Object Storage Service

制限事項

全般：

• OSS からの圧縮ファイル (GZIP、BZIP2、XZ、DEFLATE) の読み取りをサポートしているのは、Ververica Runtime (VVR) 11 以降のみです。VVR 8 は圧縮ファイルを正しく処理できません。

• 8.0.6 より前の VVR バージョンは、同じアカウント内の OSS バケットのみをサポートします。アカウント間でバケットにアクセスするには、VVR 8.0.6 以降を使用し、バケット認証を設定する必要があります。詳細については、「バケット認証の設定」をご参照ください。

• 新しいパーティションの増分読み取りはサポートされていません。

• クロスリージョンの OSS アクセスはサポートされていません。OSS バケットは Flink ワークスペースと同じリージョンにある必要があります。OSS コネクタは、単一のグローバルエンドポイントのみをサポートするファイルシステムインターフェイス上に構築されています。異なるリージョンの OSS バケットにアクセスすると、エンドポイントの不一致エラーが発生します。

シンクテーブルのみ:

OSS への書き込み時、行フォーマット (Avro、CSV、JSON、raw) はサポートされていません。詳細については、FLINK-30635 をご参照ください。

構文

CREATE TABLE OssTable (
  column_name1 INT,
  column_name2 STRING,
  ...
  datetime STRING,
  `hour` STRING
) PARTITIONED BY (datetime, `hour`) WITH (
  'connector' = 'filesystem',           -- 必須: 'filesystem' である必要があります
  'path' = 'oss://<bucket>/path',       -- 必須: OSS パスの URI
  'format' = '...',                     -- 必須: orc, parquet, avro, csv, json, または raw
  'partition.default-name' = '...',     -- オプション: パーティションフィールドが NULL または空の場合のパーティション名
  'source.monitor-interval' = '...',    -- オプション (ソースのみ): 新しいファイルをスキャンする間隔
  'auto-compaction' = '...'            -- オプション (sink のみ): 各チェックポイント後の自動コンパクションを有効にする
);

メタデータ列

ソーステーブルは、各行のファイルレベルの情報を公開するメタデータ列をサポートしています。DDL でデータ型の後に METADATA を追加してメタデータ列を定義します。

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

以下のメタデータ列が利用可能です。

WITH パラメーター

バケット認証の設定

VVR 8.0.6 以降のみがバケット認証をサポートしています。

path パラメーターを設定した後、Flink が指定された OSS パスから読み書きできるようにバケット認証を設定します。Realtime Compute 開発コンソールの [デプロイ詳細] ページの [パラメーター] タブにある [追加設定] セクションに以下を追加します。

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

<bucketName> を path パラメーターで使用したバケット名に置き換えます。

OSS-HDFS への書き込み

Realtime Compute 開発コンソールの [デプロイ詳細] ページの [パラメーター] タブにある [追加設定] セクションに以下の設定を追加します。

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

以下のいずれかの方法で OSS-HDFS エンドポイントを設定します。

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

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

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

# 追加設定
fs.oss.jindo.buckets: jindo-test,jindo-test.cn-beijing.oss-dls.aliyuncs.com

外部の Hadoop 分散ファイルシステム (HDFS) への書き込み

hdfs:// スキーマを使用するパスの場合、アクセスユーザー名を指定または切り替えるために以下を追加します。

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

例

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,                                               -- タイムスタンプ (ミリ秒)
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND    -- 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

以下の例では、StreamingFileSink と OnCheckpointRollingPolicy を使用して OSS に書き込みます。パートファイルは各チェックポイントでローリングされます。

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);

OSS-HDFS に書き込むには、[追加設定] で OSS-HDFS パラメーターも設定します。詳細については、「OSS-HDFS への書き込み」をご参照ください。

次のステップ

• サポートされるコネクタ

• Tablestore (OTS) コネクタ

• ストリーミングデータレイクハウスのための Paimon