Object Storage Service - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Realtime Compute for Apache Flink admite la lectura y escritura en Object Storage Service (OSS) mediante el conector de sistema de archivos. OSS ofrece una durabilidad de datos del 99.9999999999 % (doce nueves) y una disponibilidad del 99.995 %. Por lo tanto, es un almacenamiento confiable para canalizaciones de Flink a gran escala.

Categoría	Detalles
Tipos de tabla compatibles	Tablas de origen y de destino
Modos de ejecución	Por lotes y en streaming
Formatos de datos	ORC, Parquet, Avro, CSV, JSON y raw
Métricas de monitoreo específicas	Ninguna
Tipos de API	DataStream API y SQL
Actualización o eliminación de datos en tablas de destino	Solo inserción. No se admiten las operaciones de actualización ni eliminación.

Limitaciones

Generales:

Solo Ververica Runtime (VVR) 11 y versiones posteriores admiten la lectura de archivos comprimidos (GZIP, BZIP2, XZ, DEFLATE) desde OSS. VVR 8 no procesa correctamente los archivos comprimidos.
Las versiones de VVR anteriores a la 8.0.6 solo admiten buckets de OSS dentro de la misma cuenta. Para acceder a buckets entre cuentas distintas, use VVR 8.0.6 o posterior y configure la autenticación del bucket. Para más detalles, consulte Configure bucket authentication.
No se admite la lectura incremental de particiones nuevas.
El acceso a OSS entre regiones no es compatible con VVR 11.8 ni versiones anteriores. El bucket de OSS debe ubicarse en la misma región que el espacio de trabajo de Flink. El conector de OSS se basa en la interfaz Filesystem, que solo admite un único endpoint global. Si intenta acceder a un bucket de OSS en otra región, se generará un error por discrepancia de endpoint.
A partir de VVR 11.9, puede acceder a un bucket de OSS ubicado en una región distinta a la de su espacio de trabajo de Flink. Para ello, configure los parámetros region-id y endpoint a nivel de bucket. A continuación se muestra un ejemplo de cómo configurar estos parámetros:
```
fs.oss.bucket.<bucket-name>.region-id: cn-shanghai
fs.oss.bucket.<bucket-name>.endpoint: oss-cn-shanghai.aliyuncs.com
```

Solo para tablas de destino:

Los formatos de fila (Avro, CSV, JSON y raw) no son compatibles al escribir en OSS. Consulte FLINK-30635 para obtener más detalles.

Sintaxis

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

Columnas de metadatos

Las tablas de origen admiten columnas de metadatos que exponen información a nivel de archivo para cada fila. Defina una columna de metadatos en su DDL al agregar METADATA después del tipo de dato:

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

A continuación se muestran las columnas de metadatos disponibles:

Clave	Tipo de dato	Descripción
`file.path`	STRING NOT NULL	Ruta completa del archivo que contiene la fila.
`file.name`	STRING NOT NULL	Nombre del archivo (último elemento de la ruta).
`file.size`	BIGINT NOT NULL	Tamaño del archivo, en bytes.
`file.modification-time`	TIMESTAMP_LTZ(3) NOT NULL	Hora de la última modificación del archivo.

Parámetros WITH

Parámetros generales

Parámetro	Obligatorio	Predeterminado	Descripción
`connector`	Sí	—	Debe ser `filesystem`.
`path`	Sí	—	Ruta de OSS en formato URI, como `oss://my_bucket/my_path`. En VVR 8.0.6 y versiones posteriores, se requiere la autenticación del bucket después de establecer este parámetro. Consulte Configure bucket authentication.
`format`	Sí	—	Formato del archivo: `csv`, `json`, `avro`, `parquet`, `orc` o `raw`.

Parámetros de la tabla de origen

Parámetro	Obligatorio	Predeterminado	Descripción
`source.monitor-interval`	No	—	Intervalo para buscar archivos nuevos. Debe ser mayor que 0. Si no se establece, la ruta se analiza una sola vez y el origen queda acotado. Cada archivo se identifica por su ruta y se procesa exactamente una vez. Las rutas de los archivos procesados se almacenan en el estado y persisten entre checkpoints y savepoints. Un intervalo más corto acelera la detección de archivos, pero aumenta la frecuencia de análisis.

Parámetros de la tabla de destino

Parámetro	Obligatorio	Predeterminado	Descripción
`partition.default-name`	No	`_DEFAULT_PARTITION__`	Nombre de la partición que se usa cuando un campo de partición es NULL o una cadena vacía.
`sink.rolling-policy.file-size`	No	128 MB	Tamaño máximo del archivo parcial antes de rotarlo. Cada subtarea de destino crea como mínimo un archivo parcial por partición. Consulte Comportamiento de la política de rotación para conocer cómo interactúa esto con los formatos de archivo.
`sink.rolling-policy.rollover-interval`	No	30min	Tiempo máximo que un archivo parcial permanece abierto antes de rotarse. La frecuencia de comprobación se controla con `sink.rolling-policy.check-interval`.
`sink.rolling-policy.check-interval`	No	1min	Frecuencia con la que se comprueba si un archivo parcial debe rotarse según `sink.rolling-policy.rollover-interval`.
`auto-compaction`	No	false	Indica si se habilita la compactación automática. Los datos se escriben primero en archivos temporales. Después de cada checkpoint, los archivos temporales correspondientes se fusionan en archivos más grandes. Estos archivos temporales no son visibles antes de la fusión. Cuando esta opción está habilitada: solo se fusionan los archivos dentro de un mismo checkpoint (como mínimo uno por checkpoint); la latencia de visibilidad de los datos equivale a `checkpoint interval + compaction duration`; las ejecuciones prolongadas de compactación generan contrapresión y retrasan los checkpoints.
`compaction.file-size`	No	128 MB	Tamaño objetivo del archivo para la salida compactada. De forma predeterminada, toma el mismo valor que `sink.rolling-policy.file-size`.
`sink.partition-commit.trigger`	No	`process-time`	Momento en el que se confirma una partición. Consulte Desencadenadores de confirmación de particiones.
`sink.partition-commit.delay`	No	`0s`	Retraso mínimo antes de confirmar una partición. Establezca `1 d` para particiones diarias y `1 h` para particiones horarias.
`sink.partition-commit.watermark-time-zone`	No	`UTC`	Zona horaria para analizar una marca de agua LONG y convertirla en un TIMESTAMP durante la comparación de confirmación de particiones. Solo aplica cuando `sink.partition-commit.trigger` es `partition-time`. Use la zona horaria de la sesión cuando la marca de agua se defina en una columna TIMESTAMP_LTZ (por ejemplo, `Asia/Shanghai`). Admite nombres completos de zonas horarias (como `America/Los_Angeles`) u offsets personalizados (como `GMT-08:00`). Una configuración incorrecta retrasa las confirmaciones de partición varias horas.
`partition.time-extractor.kind`	No	`default`	Método para extraer la hora de los campos de partición. Con `default`: configure un patrón o formateador de marca de tiempo. Con `custom`: especifique una clase extractora.
`partition.time-extractor.class`	No	—	Clase que implementa la interfaz `PartitionTimeExtractor`. Es obligatoria cuando `partition.time-extractor.kind` es `custom`.
`partition.time-extractor.timestamp-pattern`	No	—	Patrón para construir una marca de tiempo a partir de los campos de partición. De forma predeterminada, el primer campo se extrae con `yyyy-MM-dd hh:mm:ss`. Ejemplos: `$dt` (un solo campo), `$year-$month-$day $hour:00:00` (varios campos), `$dt $hour:00:00` (dos campos).
`partition.time-extractor.timestamp-formatter`	No	`yyyy-MM-dd HH:mm:ss`	Formateador para convertir la cadena de marca de tiempo (definida por `partition.time-extractor.timestamp-pattern`) en una marca de tiempo. Por ejemplo, si `partition.time-extractor.timestamp-pattern` es `$year$month$day`, establezca este valor en `yyyyMMdd`. Es compatible con DateTimeFormatter de Java.
`sink.partition-commit.policy.kind`	No	—	Forma de notificar a los consumidores posteriores que una partición está lista. Con `success-file`: escribe un archivo `_SUCCESS` en el directorio de la partición. Con `custom`: usa una clase que implementa `PartitionCommitPolicy`. Se pueden combinar varias políticas.
`sink.partition-commit.policy.class`	No	—	Clase que implementa `PartitionCommitPolicy`. Es obligatoria cuando `sink.partition-commit.policy.kind` es `custom`.
`sink.partition-commit.success-file.name`	No	`_SUCCESS`	Nombre del archivo de éxito que escribe la política de confirmación `success-file`.
`sink.parallelism`	No	—	Paralelismo del operador de escritura de archivos. De forma predeterminada, toma el paralelismo del operador anterior. Debe ser mayor que 0. Cuando `auto-compaction` está habilitado, el operador de compactación también utiliza este paralelismo.
`sink.file-per-row.enabled`	No	`false`	Indica si cada registro INSERT se escribe como un archivo independiente. El contenido y la ruta relativa de cada archivo se especifican mediante `'sink.file-per-row.content-column'` y `'sink.file-per-row.path-column'`. Este parámetro es compatible con VVR 11.9 y versiones posteriores.
`sink.file-per-row.content-column`	No	—	Cuando `'sink.file-per-row.enabled'` es `true`, indica el nombre de la columna BYTES cuyos bytes se escriben como el contenido completo de cada archivo. No se admiten valores vacíos. Este parámetro es compatible con VVR 11.9 y versiones posteriores.
`sink.file-per-row.path-column`	No	—	Cuando `'sink.file-per-row.enabled'` es `true`, indica el nombre de la columna STRING cuyo valor se usa como ruta relativa de cada archivo. No se permiten rutas absolutas, esquemas URI, segmentos de ruta vacíos, `'.'`, `'..'`, barras invertidas ni caracteres NUL. Este parámetro es compatible con VVR 11.9 y versiones posteriores.

Comportamiento de la política de rotación

El comportamiento de rotación varía según el formato del archivo:

En los formatos columnares (Parquet, ORC, Avro), un archivo parcial siempre se rota durante el checkpoint, incluso si no se cumplen los criterios de la política de rotación. El tamaño del archivo y el intervalo de rotación actúan como desencadenadores adicionales entre checkpoints.

En los formatos de fila (CSV, JSON, raw), un archivo parcial se rota únicamente cuando se cumplen los criterios de la política de rotación ( sink.rolling-policy.file-size o sink.rolling-policy.rollover-interval ). Si necesita baja latencia en la visibilidad de los archivos, ajuste sink.rolling-policy.rollover-interval junto con el intervalo de sus checkpoints.

Los formatos de fila no son compatibles con las tablas de destino de OSS debido a FLINK-30635 . El comportamiento descrito aplicará si se añade compatibilidad con formatos de fila en una versión futura.

Desencadenadores de confirmación de particiones

Existen dos tipos de desencadenadores para sink.partition-commit.trigger:

process-time (predeterminado): confirma una partición cuando la hora actual del sistema supera la hora de creación de la partición más sink.partition-commit.delay. No requiere una marca de agua ni un extractor de tiempo de partición. Es más general, pero menos preciso, ya que los retrasos o fallos en los datos provocan confirmaciones prematuras.
partition-time: confirma una partición cuando la marca de agua supera la hora de creación de la partición más sink.partition-commit.delay. Requiere la generación de marcas de agua y particiones basadas en tiempo (horarias, diarias, etc.).

Configurar la autenticación del bucket

Solo VVR 8.0.6 y versiones posteriores admiten la autenticación de buckets.

Después de establecer el parámetro path, configure la autenticación del bucket para que Flink lea y escriba en la ruta de OSS especificada. Agregue la siguiente configuración en la sección Additional Configurations de la pestaña Parameters en la página Deployment Details de la consola de desarrollo de Realtime Compute:

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

Reemplace <bucketName> con el nombre del bucket utilizado en el parámetro path.

Elemento de configuración	Descripción
`fs.oss.bucket.<bucketName>.accessKeyId`	AccessKey ID del bucket. Utilice un AccessKey existente o cree uno nuevo. Consulte Create an AccessKey.
`fs.oss.bucket.<bucketName>.accessKeySecret`	AccessKey Secret del bucket.

Importante

El AccessKey Secret se muestra una única vez al momento de crearlo. Guárdelo de forma segura.

Escribir en OSS-HDFS

Agregue la siguiente configuración en la sección Other Configuration de la pestaña Parameters en la página Deployment Details de la consola de desarrollo de Realtime Compute:

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

Elemento de configuración	Descripción
`fs.oss.jindo.buckets`	Nombres de los buckets de OSS-HDFS, separados por punto y coma. Cuando Flink escribe en una ruta de OSS y el bucket correspondiente aparece en esta lista, los datos se escriben en el servicio OSS-HDFS.
`fs.oss.jindo.accessKeyId`	AccessKey ID. Consulte Create an AccessKey.
`fs.oss.jindo.accessKeySecret`	AccessKey Secret.

Importante

El AccessKey Secret se muestra una única vez al momento de crearlo. Guárdelo de forma segura.

Configure el endpoint de OSS-HDFS mediante uno de los siguientes métodos:

Configuración por parámetros

Agregue el endpoint en Additional Configurations:

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

Configuración por ruta

Incruste el endpoint directamente en la ruta de OSS:

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

Al usar este método, fs.oss.jindo.buckets debe incluir <bucket-name>.<oss-hdfs-endpoint>.

Por ejemplo, si el nombre del bucket es jindo-test y el endpoint es cn-beijing.oss-dls.aliyuncs.com:

# 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

Escritura en un sistema de archivos distribuido Hadoop (HDFS) externo

Para las rutas que utilizan el esquema hdfs://, agregue lo siguiente para especificar o cambiar el nombre de usuario de acceso:

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

Ejemplos

Leer desde OSS (tabla de origen)

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;

Escribir en OSS (tabla de destino)

Escribir en una tabla particionada

Este ejemplo transmite datos desde un origen datagen, los particiona por fecha y hora, y confirma las particiones mediante el desencadenador 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;

Escribir en una tabla no particionada

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

Importante

Para utilizar DataStream API, primero configure el conector de DataStream. Consulte Use a DataStream connector.

El siguiente ejemplo utiliza StreamingFileSink con OnCheckpointRollingPolicy para escribir en OSS. Los archivos parciales se rotan en cada 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);

Para escribir en OSS-HDFS, configure también los parámetros correspondientes en Additional Configurations. Consulte Write to OSS-HDFS.

Realtime Compute for Apache Flink:Object Storage Service

Limitaciones

Sintaxis

Columnas de metadatos

Parámetros WITH

Parámetros generales

Parámetros de la tabla de origen

Parámetros de la tabla de destino

Comportamiento de la política de rotación

Desencadenadores de confirmación de particiones

Configurar la autenticación del bucket

Escribir en OSS-HDFS

Configuración por parámetros

Configuración por ruta

Ejemplos

Leer desde OSS (tabla de origen)

Escribir en OSS (tabla de destino)

DataStream API

Pasos siguientes