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 roulement

Le comportement de roulement varie selon le format de fichier :

Pour les formats columnaires (Parquet, ORC, Avro), un fichier partiel est toujours roulé au moment du point de contrôle, même si les critères de la politique de roulement ne sont pas remplis. La taille du fichier et l'intervalle de roulement s'appliquent comme déclencheurs supplémentaires entre les points de contrôle.

Pour les formats de ligne (CSV, JSON, brut), un fichier partiel est roulé uniquement lorsque les critères de la politique de roulement ( 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 point de contrôle.

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

Déclencheurs de validation 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 ou les pannes de données 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 les données d'une source datagen, les partitionne par date et heure, et valide les partitions à l'aide du 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);