Realtime Compute for Apache Flink:Object Storage Service

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

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

Paramètres généraux

Paramètres de la table source

Paramètres de la table puits

Comportement de la politique de rotation

Le comportement de rotation varie selon le format de fichier :

Pour les formats columnaires (Parquet, ORC, Avro), un fichier partiel subit toujours une rotation au moment du checkpoint, même si les critères de la politique de rotation ne sont pas remplis. La taille du fichier et l'intervalle de rotation s'appliquent comme déclencheurs supplémentaires entre les checkpoints.

Pour les formats de ligne (CSV, JSON, raw), un fichier partiel subit une rotation uniquement lorsque les critères de la politique de rotation ( sink.rolling-policy.file-size ou sink.rolling-policy.rollover-interval ) sont remplis. Si vous avez besoin d'une faible latence de visibilité des fichiers, ajustez sink.rolling-policy.rollover-interval conjointement avec votre intervalle de checkpoint.

Les formats de ligne ne sont pas pris en charge pour les tables puits OSS en raison de FLINK-30635 . Le comportement décrit ci-dessus s'appliquera si la prise en charge des formats de ligne est ajoutée dans une version future.

Déclencheurs de commit de partition

Deux types de déclencheurs sont disponibles pour sink.partition-commit.trigger :

• process-time (par défaut) : Valide une partition lorsque l'heure système actuelle dépasse l'heure de création de la partition plus sink.partition-commit.delay . Ne nécessite ni watermark ni extracteur de temps de partition. Plus général, mais moins précis : les retards de données ou les pannes peuvent entraîner des validations prématurées.

• partition-time : Valide une partition lorsque le watermark dépasse l'heure de création de la partition plus sink.partition-commit.delay . Nécessite la génération de watermarks et des partitions basées sur le temps (horaire, quotidien, etc.).

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

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

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

Lire depuis OSS (table source)

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;

Écrire dans OSS (table puits)

Écrire dans une table partitionnée

Cet exemple diffuse des données depuis une source datagen , les partitionne par date et heure, et valide les partitions en utilisant le déclencheur partition-time :

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;

Écrire dans une table non partitionnée

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;

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