Todos los productos
Search

Este producto

    Tablestore:Uso de Realtime Compute for Apache Flink para procesar datos de Tablestore

    Centro de documentos

    Tablestore:Uso de Realtime Compute for Apache Flink para procesar datos de Tablestore

    Última actualización: Sep 03, 2026

    Este tema describe cómo procesar datos de Tablestore mediante Realtime Compute for Apache Flink. Una tabla de datos o una tabla de series temporales en Tablestore se puede utilizar como tabla de origen o tabla de resultados en el procesamiento de datos con Realtime Compute for Apache Flink.

    Requisitos previos

    Desarrollo de un trabajo de cómputo en tiempo real

    Paso 1: Crear un borrador SQL

    1. Vaya a la página para crear un borrador.

      1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

      2. En la columna Actions del espacio de trabajo deseado, haga clic en Console.

      3. En el panel de navegación izquierdo, haga clic en Development > ETL.

    2. Haga clic en New. En el cuadro de diálogo New Draft, seleccione Blank Stream Draft y haga clic en Next.

      Nota

      Realtime Compute for Apache Flink proporciona diversas plantillas de código y admite la sincronización de datos. Cada plantilla de código es adecuada para escenarios específicos y ofrece ejemplos de código e instrucciones. Puede hacer clic en una plantilla para conocer las funciones y la sintaxis relacionada de Realtime Compute for Apache Flink, e implementar su lógica de negocio. Para obtener más información, consulte Code templates y Data synchronization templates.

    3. Introduzca la Job Information.

      Parámetro

      Descripción

      Ejemplo

      File Name

      Nombre del borrador que desea crear.

      Nota

      El nombre del borrador debe ser único en el proyecto actual.

      flink-test

      Location

      Carpeta donde se guarda el archivo de código del borrador.

      También puede hacer clic en el icono 新建文件夹 a la derecha de una carpeta existente para crear una subcarpeta.

      Draft

      Engine version

      Versión del motor Flink que desea utilizar para el borrador actual. Para obtener más información sobre las versiones del motor, consulte Release notes y Engine version.

      vvr-8.0.10-flink-1.17

    4. Haga clic en Create.

    Paso 2: Escribir el borrador SQL

    Nota

    En el ejemplo de este paso, se escribe código para sincronizar datos de una tabla de datos a otra. Para obtener información sobre más ejemplos de sentencias SQL, consulte Sample SQL statements.

    1. Cree una tabla temporal para la tabla de origen y la tabla de resultados.

      Para obtener más información, consulte Appendix 1: Tablestore connector.

      -- Create a temporary table named tablestore_stream for the source table.
CREATE TEMPORARY TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector' = 'ots', -- Specify the connector type of the source table. The value is ots and cannot be changed. 
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com', -- Specify the virtual private cloud (VPC) endpoint of the Tablestore instance. 
    'instanceName' = 'xxx', -- Specify the name of the Tablestore instance. 
    'tableName' = 'flink_source_table', -- Specify the name of the source table. 
    'tunnelName' = 'flink_source_tunnel', -- Specify the name of the tunnel that is created for the source table. 
    'accessId' = 'xxxxxxxxxxx', -- Specify the AccessKey ID of the Alibaba Cloud account or RAM user. 
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx', -- Specify the AccessKey secret of the Alibaba Cloud account or RAM user. 
    'ignoreDelete' = 'false' -- Specify whether to ignore the real-time data that is generated by delete operations. In this example, this parameter is set to false. 
);

-- Create a temporary table named tablestore_sink for the result table.
CREATE TEMPORARY TABLE tablestore_sink(
   `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    PRIMARY KEY (`order`,orderid) NOT ENFORCED -- Specify the primary key. 
) WITH (
    'connector' = 'ots', -- Specify the connector type of the result table. The value is ots and cannot be changed. 
    'endPoint'='https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com', -- Specify the VPC endpoint of the Tablestore instance. 
    'instanceName' = 'xxx', -- Specify the name of the Tablestore instance. 
    'tableName' = 'flink_sink_table', -- Specify the name of the result table. 
    'accessId' = 'xxxxxxxxxxx',  -- Specify the AccessKey ID of the Alibaba Cloud account or RAM user. 
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx', -- Specify the AccessKey secret of the Alibaba Cloud account or RAM user. 
    'valueColumns'='customerid,customername' --Specify the names of the columns that you want to insert to the result table. 
);

    2. Escriba la lógica del borrador.

      La siguiente sentencia SQL de ejemplo muestra cómo insertar datos de la tabla de origen en la tabla de resultados:

      -- Insert data from the source table into the result table.
INSERT INTO tablestore_sink
SELECT `order`, orderid, customerid, customername FROM tablestore_stream;

    Paso 3: (Opcional) Consultar la información de configuración

    En la pestaña derecha del editor SQL, puede consultar las configuraciones o configurar los parámetros. La siguiente tabla describe los parámetros.

    Nombre de la pestaña

    Descripción

    Configurations

    • Engine version: Versión del motor Flink para el borrador.

    • Additional dependencies: Dependencias adicionales necesarias para el trabajo, como UDF.

      Puede descargar las dependencias de Ververica Runtime (VVR), cargarlas en la página de archivos de recursos y luego seleccionar los archivos cargados como additional dependencies. Para obtener más información, consulte Appendix 2: Configure VVR dependencies.

    Structure

    • Data flow diagram: Visualiza el flujo de datos del trabajo.

    • Tree structure diagram: Visualiza el linaje de datos del trabajo.

    Versions

    Muestra el historial de versiones del borrador. Para obtener detalles sobre las funciones en la columna Actions, consulte Manage job versions.

    Paso 4: (Opcional) Realizar una comprobación de sintaxis

    La validación comprueba la semántica SQL del trabajo, la conectividad de red y los metadatos de las tablas. También puede hacer clic en SQL Advice en el área de resultados para ver posibles riesgos de SQL y sugerencias de optimización.

    1. En la esquina superior derecha del editor SQL, haga clic en Validate.

    2. En el cuadro de diálogo Validation, haga clic en Confirm.

    Paso 5: (Opcional) Depurar el borrador

    Puede utilizar la función de depuración para simular la ejecución del despliegue, comprobar las salidas y verificar la lógica de negocio de las sentencias SELECT e INSERT. Esta función mejora la eficiencia del desarrollo y reduce los riesgos de una calidad de datos deficiente.

    1. En la esquina superior derecha del editor SQL, haga clic en Debug.

    2. En el cuadro de diálogo Debug, seleccione un clúster de sesión y haga clic en Next.

      Si no hay ningún clúster disponible, cree un clúster de sesión. Asegúrese de que el clúster de sesión utilice la misma versión del motor que el borrador SQL y de que el clúster de sesión esté en estado Running. Para obtener más información, consulte Create a session cluster.

    3. Configure los datos de depuración.

      • Si utiliza datos en línea, omita esta operación.

      • Si desea utilizar datos de depuración, haga clic en Download Debugging Data Template, complete la plantilla con sus datos y luego cargue el archivo. Para obtener más información, consulte Debugging a job.

    4. Después de configurar los datos, haga clic en OK.

    Paso 6: Desplegar el borrador

    En la esquina superior derecha del editor SQL, haga clic en Deploy. En el cuadro de diálogo Deploy New Version, configure los parámetros de despliegue y haga clic en OK.

    Nota

    Los clústeres de sesión son adecuados para entornos que no son de producción, como entornos de desarrollo y prueba. Puede desplegar o depurar borradores en un clúster de sesión para mejorar la utilización de recursos de un JobManager y acelerar el inicio del despliegue. Sin embargo, se recomienda no desplegar borradores para el entorno de producción en clústeres de sesión; de lo contrario, pueden producirse problemas de estabilidad.

    Paso 7: Iniciar el despliegue del borrador y consultar el resultado del cómputo

    1. En el panel de navegación izquierdo, haga clic en O&M > Deployments.

    2. En la columna Actions del despliegue deseado, haga clic en Start.

      Seleccione Start with no state y haga clic en Start. Un estado Running indica que el despliegue funciona correctamente. Para obtener más información sobre los parámetros de inicio, consulte Start a job.

      Nota

      • Se recomienda configurar dos núcleos de CPU y 4 GB de memoria para cada TaskManager en Realtime Compute for Apache Flink con el fin de maximizar las capacidades de cómputo de cada TaskManager. Un TaskManager puede escribir 10,000 filas por segundo.

      • Si el número de particiones en la tabla de origen es grande, se recomienda establecer la concurrencia en menos de 16 en Realtime Compute for Apache Flink. La tasa de escritura aumenta linealmente con la concurrencia.

    3. En la página Deployments, consulte el resultado del cómputo.

      1. En la página O&M > Deployments, haga clic en el nombre del despliegue deseado.

      2. En la pestaña Job logs, haga clic en la pestaña Running task managers y luego haga clic en la tarea deseada en la columna Path,ID.

      3. Haga clic en Logs para consultar la información de registro.

    4. (Opcional) Cancelar un despliegue.

      Si modifica el código SQL de un despliegue, agrega o elimina parámetros de la cláusula WITH, o cambia la versión de un despliegue, debe desplegar el borrador del despliegue, cancelar el despliegue y luego iniciar el despliegue para que los cambios surtan efecto. Si un despliegue falla y no puede reutilizar los datos de estado para recuperarse, o si desea actualizar la configuración de parámetros que no surten efecto dinámicamente, debe cancelar y luego reiniciar el despliegue. Para obtener más información sobre cómo cancelar un despliegue, consulte Cancel a deployment.

    Apéndices

    Apéndice 1: Conector de Tablestore

    Realtime Compute for Apache Flink proporciona un Tablestore connector integrado para leer, escribir y sincronizar datos de Tablestore.

    Tabla de origen

    Sintaxis DDL

    Tabla de datos

    El siguiente código de ejemplo muestra la sentencia DDL para crear una tabla temporal para la tabla de origen:

    -- Create a temporary table named tablestore_stream for the source table.
CREATE TEMPORARY TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false'
);

    Tabla de series temporales

    El siguiente código de ejemplo muestra la sentencia DDL para crear una tabla temporal para la tabla de origen:

    -- Create a temporary table named tablestore_stream for the source table.
CREATE TEMPORARY TABLE tablestore_stream(
    _m_name STRING,
    _data_source STRING,
    _tags STRING,
    _time BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx'
);

    Puede leer campos de metadatos del Tunnel Service, como OtsRecordType y OtsRecordTimestamp, como columnas normales en la tabla de origen. La siguiente tabla describe estos campos.

    Campo

    Campo asignado en Realtime Compute for Apache Flink

    Descripción

    OtsRecordType

    type

    Tipo de operación.

    OtsRecordTimestamp

    timestamp

    Hora de la operación de datos. Unidad: microsegundos.

    Nota

    Si desea que Realtime Compute for Apache Flink lea todos los datos, establezca este campo en 0.

    Parámetro en la cláusula WITH

    Parámetro

    Tabla aplicable

    Obligatorio

    Descripción

    connector

    General

    Sí

    Tipo de conector de la tabla de origen. El valor es ots y no se puede cambiar.

    endPoint

    General

    Sí

    Endpoint de la instancia de Tablestore. Debe utilizar un endpoint de VPC. Para obtener más información, consulte Endpoints.

    instanceName

    General

    Sí

    Nombre de la instancia de Tablestore.

    tableName

    General

    Sí

    Nombre de la tabla de origen en Tablestore.

    tunnelName

    General

    Sí

    Nombre del túnel para la tabla de origen en Tablestore. Para obtener información sobre cómo crear un túnel, consulte Create a tunnel.

    accessId

    General

    Sí

    Par de AccessKey (AccessKey ID y AccessKey secret) de la cuenta de Alibaba Cloud o del usuario de RAM.

    Importante

    Para proteger su par de AccessKey, se recomienda utilizar variables para especificar el par de AccessKey. Para obtener más información, consulte Manage variables.

    accessKey

    General

    Sí

    connectTimeout

    General

    No

    Tiempo de espera para que el conector de Tablestore se conecte a Tablestore. Unidad: milisegundos. Valor predeterminado: 30000.

    socketTimeout

    General

    No

    Tiempo de espera del socket para que el conector de Tablestore se conecte a Tablestore. Unidad: milisegundos. Valor predeterminado: 30000.

    ioThreadCount

    General

    No

    Número de hilos de E/S. Valor predeterminado: 4.

    callbackThreadPoolSize

    General

    No

    Tamaño del pool de hilos de devolución de llamada. Valor predeterminado: 4.

    ignoreDelete

    Tabla de datos

    No

    Especifica si se ignoran los datos en tiempo real generados por operaciones de eliminación. Valor predeterminado: false, lo que indica que los datos en tiempo real generados por operaciones de eliminación no se ignoran.

    skipInvalidData

    General

    No

    Especifica si se ignoran los datos incorrectos. Valor predeterminado: false, lo que indica que los datos incorrectos no se ignoran. Si no se ignoran los datos incorrectos, el sistema notifica un error al procesarlos.

    Importante

    Solo Realtime Compute for Apache Flink que utiliza VVR 8.0.4 o posterior admite este parámetro.

    retryStrategy

    General

    No

    Directiva de reintento. Valores válidos:

    • TIME: El sistema reintenta continuamente hasta que finaliza el período de tiempo de espera especificado por el parámetro retryTimeoutMs. Este es el valor predeterminado.

    • COUNT: El sistema reintenta continuamente hasta alcanzar el número máximo de reintentos especificado por el parámetro retryCount.

    retryCount

    General

    No

    Número máximo de reintentos. Si establece el parámetro retryStrategy en COUNT, puede configurar este parámetro. Valor predeterminado: 3.

    retryTimeoutMs

    General

    No

    Período de tiempo de espera para los reintentos. Unidad: milisegundos. Si establece el parámetro retryStrategy en TIME, puede configurar este parámetro. Valor predeterminado: 180000.

    streamOriginColumnMapping

    General

    No

    Asignación entre los nombres de columna en la tabla de origen y los nombres de columna en la tabla temporal.

    Nota

    Utilice dos puntos (:) para separar un nombre de columna original y un nombre de columna real. Utilice comas (,) para separar varias asignaciones. Ejemplo: origin_col1:col1,origin_col2:col2.

    outputSpecificRowType

    General

    No

    Especifica si se transmite un tipo de fila específico. Valores válidos:

    • false: no transmite un tipo de fila específico. El tipo de fila de todos los datos es INSERT. Este es el valor predeterminado.

    • true: transmite un tipo de fila específico. El tipo de fila de los datos puede ser INSERT, DELETE o UPDATE_AFTER.

    Asignaciones de tipos de datos

    Tipo de datos de campo en Tablestore

    Tipo de datos de campo en Realtime Compute for Apache Flink

    INTEGER

    BIGINT

    STRING

    STRING

    BOOLEAN

    BOOLEAN

    DOUBLE

    DOUBLE

    BINARY

    BINARY

    Tabla de resultados

    Sintaxis DDL

    Tabla de datos

    El siguiente código de ejemplo muestra la sentencia DDL para crear una tabla temporal para la tabla de resultados:

    -- Create a temporary table named tablestore_sink for the result table.
CREATE TEMPORARY TABLE tablestore_sink(
   `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    PRIMARY KEY (`order`,orderid) NOT ENFORCED
) WITH (
    'connector' = 'ots',
    'endPoint'='https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_sink_table',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'valueColumns'='customerid,customername'
);
    Nota

    Debe especificar el esquema de la clave principal y al menos una columna de atributo para la tabla de resultados de Tablestore. Los datos de salida se anexan a la tabla de resultados de Tablestore para actualizar los datos de la tabla.

    Tabla de series temporales

    Una tabla de resultados de series temporales requiere cuatro claves principales: _m_name, _data_source, _tags y _time. Puede especificar estas claves principales de tres maneras: mediante parámetros WITH, mediante la definición de clave principal de la tabla de resultados o mediante una clave principal con formato Map. Al definir la columna _tags, el método de parámetro WITH tiene la mayor prioridad, seguido por el formato Map y la definición de clave principal de la tabla de resultados.

    Uso de los parámetros en la cláusula WITH

    El siguiente código de ejemplo muestra cómo utilizar los parámetros en la cláusula WITH para definir la sintaxis DDL:

    -- Create a temporary table named tablestore_sink for the result table.
CREATE TEMPORARY TABLE tablestore_sink(
    measurement STRING,
    datasource STRING,
    tag_a STRING,
    `time` BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING,
    tag_b STRING,
    tag_c STRING,
    tag_d STRING,
    tag_e STRING,
    tag_f STRING,
    PRIMARY KEY(measurement, datasource, tag_a, `time`) NOT ENFORCED
) 
WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_sink_table',
    'accessId' ='xxxxxxxxxxx',
    'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'storageType' = 'TIMESERIES',
    'timeseriesSchema' = '{"measurement":"_m_name", "datasource":"_data_source", "tag_a":"_tags", "tag_b":"_tags", "tag_c":"_tags", "tag_d":"_tags", "tag_e":"_tags", "tag_f":"_tags", "time":"_time"}'
);

-- Insert data from the source table into the result table.
INSERT INTO tablestore_sink
    select 
    measurement,
    datasource,
    tag_a,
    `time`,
    binary_value,
    bool_value,
    double_value,
    long_value,
    string_value,
    tag_b,
    tag_c,
    tag_d,
    tag_e,
    tag_f
    from tablestore_stream;

    Uso de la clave principal en formato Map

    El siguiente código de ejemplo muestra cómo utilizar la clave principal en formato Map para definir la sintaxis DDL:

    Nota

    Tablestore proporciona el tipo de datos Map de Flink para facilitar la generación de la columna _tags de la tabla de series temporales en el modelo TimeSeries. El tipo de datos Map admite operaciones de asignación, como cambio de nombre de columnas y funciones simples. Cuando utilice Map, asegúrese de que la columna de clave principal _tags se encuentre en la tercera posición.

    -- Create a temporary table named tablestore_sink for the result table.
CREATE TEMPORARY TABLE tablestore_sink(
    measurement STRING,
    datasource STRING,
    tags Map<String, String>, 
    `time` BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING,
    PRIMARY KEY(measurement, datasource, tags, `time`) NOT ENFORCED
)
WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_sink_table',
    'accessId' ='xxxxxxxxxxx',
    'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'storageType' = 'TIMESERIES'
);

-- Insert data from the source table into the result table.
INSERT INTO tablestore_sink
    select 
    measurement,
    datasource,
    MAP[`tag_a`, `tag_b`, `tag_c`, `tag_d`, `tag_e`, `tag_f`] AS tags,
    `time`,
    binary_value,
    bool_value,
    double_value,
    long_value,
    string_value
    from timeseries_source;

    Uso de la clave principal de la tabla SINK

    El siguiente código de ejemplo muestra cómo utilizar la clave principal de la tabla SINK para definir la sintaxis DDL. La primera columna de clave principal es la columna _m_name, que especifica el nombre de la medición. La segunda columna de clave principal es la columna _data_source, que especifica el origen de datos. La última columna de clave principal es la columna _time, que especifica la marca de tiempo. La columna de clave principal intermedia es la columna _tags, que especifica las etiquetas de la serie temporal.

    -- Create a temporary table named tablestore_sink for the result table.
CREATE TEMPORARY TABLE tablestore_sink(
    measurement STRING,
    datasource STRING,
    tag_a STRING,
    `time` BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING,
    tag_b STRING,
    tag_c STRING,
    tag_d STRING,
    tag_e STRING,
    tag_f STRING
    PRIMARY KEY(measurement, datasource, tag_a, tag_b, tag_c, tag_d, tag_e, tag_f, `time`) NOT ENFORCED
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_sink_table',
    'accessId' ='xxxxxxxxxxx',
    'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'storageType' = 'TIMESERIES'
);

-- Insert data from the source table into the result table.
INSERT INTO tablestore_sink
    select 
    measurement,
    datasource,
    tag_a,
    tag_b,
    tag_c,
    tag_d,
    tag_e,
    tag_f,
    `time`,
    binary_value,
    bool_value,
    double_value,
    long_value,
    string_value
    from timeseries_source;
    Parámetro en la cláusula WITH

    Parámetro

    Tabla aplicable

    Obligatorio

    Descripción

    connector

    General

    Sí

    Tipo de conector de la tabla de resultados. El valor es ots y no se puede modificar.

    endPoint

    General

    Sí

    Endpoint de la instancia de Tablestore. Debe usar un endpoint de VPC. Para obtener más información, consulte Endpoints.

    instanceName

    General

    Sí

    Nombre de la instancia de Tablestore.

    tableName

    General

    Sí

    Nombre de la tabla de series temporales en Tablestore.

    accessId

    General

    Sí

    Par de AccessKey (AccessKey ID y AccessKey secret) de la cuenta de Alibaba Cloud o del usuario de RAM.

    Importante

    Para proteger su par de AccessKey, se recomienda utilizar variables para especificarlo. Para obtener más información, consulte Manage variables.

    accessKey

    General

    Sí

    valueColumns

    Tabla de datos

    Sí

    Nombres de las columnas donde se escribirán los datos. Separe varios nombres de columna con comas (,). Ejemplo: ID,NAME.

    storageType

    General

    No

    Importante

    Si utiliza una tabla de series temporales como tabla de resultados, establezca este parámetro en TIMESERIES.

    Tipo de tabla. Valores válidos:

    • WIDE_COLUMN: tabla de datos. Este es el valor predeterminado.

    • TIMESERIES: tabla de series temporales.

    timeseriesSchema

    Tabla de series temporales

    No

    Importante

    Cuando utilice una tabla de series temporales como tabla de resultados, debe configurar este parámetro si usa los parámetros de la cláusula WITH para especificar la clave principal de dicha tabla.

    Columnas que desea definir como clave principal de la tabla de series temporales.

    • Especifique la clave principal de la tabla de series temporales como pares clave-valor en formato JSON. Ejemplo: {"measurement":"_m_name","datasource":"_data_source","tag_a":"_tags","tag_b":"_tags","tag_c":"_tags","tag_d":"_tags","tag_e":"_tags","tag_f":"_tags","time":"_time"}.

    • Los tipos de las columnas de clave principal que especifique deben coincidir con los tipos de las columnas de clave principal en la tabla de series temporales. La columna de clave principal de etiquetas puede constar de varias columnas.

    connectTimeout

    General

    No

    Tiempo de espera para que el conector de Tablestore se conecte a Tablestore. Unidad: milisegundos. Valor predeterminado: 30000.

    socketTimeout

    General

    No

    Tiempo de espera del socket para que el conector de Tablestore se conecte a Tablestore. Unidad: milisegundos. Valor predeterminado: 30000.

    ioThreadCount

    General

    No

    Número de hilos de E/S. Valor predeterminado: 4.

    callbackThreadPoolSize

    General

    No

    Tamaño del pool de hilos de devolución de llamada. Valor predeterminado: 4.

    retryIntervalMs

    General

    No

    Intervalo entre reintentos. Unidad: milisegundos. Valor predeterminado: 1000.

    maxRetryTimes

    General

    No

    Número máximo de reintentos. Valor predeterminado: 10.

    bufferSize

    General

    No

    Número máximo de registros de datos que se pueden almacenar en el búfer antes de escribirlos en la tabla de resultados. Valor predeterminado: 5000, lo que indica que los datos se escriben en la tabla de resultados cuando el número de registros en el búfer alcanza los 5,000.

    batchWriteTimeoutMs

    General

    No

    Tiempo de espera de escritura. Unidad: milisegundos. Valor predeterminado: 5000, lo que indica que todos los datos del búfer se escriben en la tabla de resultados si el número de registros no alcanza el valor especificado por el parámetro bufferSize dentro de 5,000 milisegundos.

    batchSize

    General

    No

    Número de registros de datos que se pueden escribir simultáneamente en la tabla de resultados. Valor predeterminado: 100. Valor máximo: 200.

    ignoreDelete

    General

    No

    Indica si se deben ignorar los datos en tiempo real generados por operaciones de eliminación. Valor predeterminado: false, lo que significa que no se ignoran los datos en tiempo real generados por operaciones de eliminación.

    Importante

    Si utiliza una tabla de datos como tabla de origen, puede configurar este parámetro según sus requisitos empresariales.

    autoIncrementKey

    Tabla de datos

    No

    Nombre de la columna de clave principal de incremento automático de la tabla de resultados, cuando dicha tabla contiene una columna de este tipo. Si la tabla de resultados no tiene una columna de clave principal de incremento automático, no es necesario configurar este parámetro.

    Importante

    Solo Realtime Compute for Apache Flink con VVR 8.0.4 o posterior admite este parámetro.

    overwriteMode

    General

    No

    Modo de sobrescritura de datos. Valores válidos:

    • PUT: Los datos se escriben en la tabla de Tablestore en modo PUT. Este es el valor predeterminado.

    • UPDATE: Los datos se escriben en la tabla de Tablestore en modo UPDATE.

    Nota

    Solo se admite el modo UPDATE en el modo de columna dinámica.

    defaultTimestampInMillisecond

    General

    No

    Marca de tiempo predeterminada que se utiliza para escribir datos en la tabla de Tablestore. Si deja este parámetro vacío, se utilizará la marca de tiempo actual del sistema.

    dynamicColumnSink

    General

    No

    Indica si se debe habilitar el modo de columna dinámica. Valor predeterminado: false, lo que significa que el modo de columna dinámica está deshabilitado.

    Nota

    • El modo de columna dinámica es adecuado para escenarios en los que no se especifican columnas para una tabla y las columnas de datos se insertan según el estado del despliegue. Debe especificar las primeras columnas como columnas de clave principal en la sentencia de creación de la tabla. El valor de la penúltima columna se usa como variable de nombre de columna, el valor de la última columna se usa como valor de dicha variable, y el tipo de dato de la penúltima columna debe ser String.

    • Si habilita el modo de columna dinámica, la función de columna de clave principal de incremento automático no es compatible y debe establecer el parámetro overwriteMode en UPDATE.

    checkSinkTableMeta

    General

    No

    Indica si se deben comprobar los metadatos de la tabla de resultados. Valor predeterminado: true, lo que significa que el sistema comprueba si las columnas de clave principal de la tabla de Tablestore coinciden con las columnas de clave principal especificadas en la sentencia de creación de la tabla.

    enableRequestCompression

    General

    No

    Indica si se debe habilitar la compresión de datos durante la escritura. Valor predeterminado: false, lo que significa que la compresión de datos está deshabilitada durante la escritura.

    Asignaciones de tipos de datos

    Tipo de dato del campo en Realtime Compute for Apache Flink

    Tipo de dato del campo en Tablestore

    BINARY

    BINARY

    VARBINARY

    BINARY

    CHAR

    STRING

    VARCHAR

    STRING

    TINYINT

    INTEGER

    SMALLINT

    INTEGER

    INTEGER

    INTEGER

    BIGINT

    INTEGER

    FLOAT

    DOUBLE

    DOUBLE

    DOUBLE

    BOOLEAN

    BOOLEAN

    Sentencias SQL de ejemplo

    Sincronizar datos de la tabla de origen a la tabla de resultados

    Sincronizar datos de una tabla de datos a una tabla de series temporales

    Lea datos de una tabla de datos llamada flink_source_table y escríbalos en una tabla de series temporales llamada flink_sink_table.

    Sentencia SQL de ejemplo:

    -- Create a temporary table named tablestore_stream for the source table.
CREATE TEMPORARY TABLE tablestore_stream(
    measurement STRING,
    datasource STRING,
    tag_a STRING,
    `time` BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING,
    tag_b STRING,
    tag_c STRING,
    tag_d STRING,
    tag_e STRING,
    tag_f STRING
) 
WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false' 
);

-- Create a temporary table named tablestore_sink for the result table by using the parameters in the WITH clause.
CREATE TEMPORARY TABLE tablestore_sink(
     measurement STRING,
     datasource STRING,
     tag_a STRING,
     `time` BIGINT,
     binary_value BINARY,
     bool_value BOOLEAN,
     double_value DOUBLE,
     long_value BIGINT,
     string_value STRING,
     tag_b STRING,
     tag_c STRING,
     tag_d STRING,
     tag_e STRING,
     tag_f STRING,
     PRIMARY KEY(measurement, datasource, tag_a, `time`) NOT ENFORCED
 ) WITH (
     'connector' = 'ots',
     'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
     'instanceName' = 'xxx',
     'tableName' = 'flink_sink_table',
     'accessId' = 'xxxxxxxxxxx',
     'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
     'storageType' = 'TIMESERIES',
     'timeseriesSchema' = '{"measurement":"_m_name","datasource":"_data_source","tag_a":"_tags","tag_b":"_tags","tag_c":"_tags","tag_d":"_tags","tag_e":"_tags","tag_f":"_tags","time":"_time"}'
 );
 
-- Insert data from the source table into the result table.
INSERT INTO tablestore_sink
    select 
    measurement,
    datasource,
    tag_a,
    `time`,
    binary_value,
    bool_value,
    double_value,
    long_value,
    string_value,
    tag_b,
    tag_c,
    tag_d,
    tag_e,
    tag_f
    from tablestore_stream;

    Sincronizar datos de una tabla de series temporales a una tabla de datos

    Lea datos de una tabla de series temporales llamada flink_source_table y escríbalos en una tabla de datos llamada flink_sink_table.

    Sentencia SQL de ejemplo:

    -- Create a temporary table named tablestore_stream for the source table.
CREATE TEMPORARY TABLE tablestore_stream(
    _m_name STRING,
    _data_source STRING,
    _tags STRING,
    _time BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx'
);

-- Create a temporary table named print_table for the result table. 
CREATE TEMPORARY TABLE tablestore_target(
    measurement STRING,
    datasource STRING,
    tags STRING,
    `time` BIGINT,
    binary_value BINARY,
    bool_value BOOLEAN,
    double_value DOUBLE,
    long_value BIGINT,
    string_value STRING,
    PRIMARY KEY (measurement,datasource, tags, `time`) NOT ENFORCED
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_sink_table',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'valueColumns'='binary_value,bool_value,double_value,long_value,string_value'
);

-- Insert data from the source table into the result table.
INSERT INTO tablestore_target
SELECT
    _m_name,
    _data_source,
    _tags,
    _time,
    binary_value,
    bool_value,
    double_value,
    long_value,
    string_value
    from tablestore_stream;

    Leer datos de la tabla de origen e imprimirlos en la consola de Tablestore

    Lea datos de la tabla de origen llamada flink_source_table en lotes. Puede utilizar la función de depuración de despliegues para simular la ejecución de un despliegue. El resultado de la depuración se muestra en la parte inferior del editor SQL.

    Sentencia SQL de ejemplo:

    -- Create a temporary table named tablestore_stream for the source data table.
CREATE TEMPORARY TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false' 
);

-- Read data from the source table.
SELECT * FROM tablestore_stream LIMIT 100;

    Leer datos de la tabla de origen e imprimirlos en el registro de TaskManager

    Lea datos de la tabla de origen llamada flink_source_table e imprima los resultados en el registro de TaskManager mediante el conector Print.

    Sentencia SQL de ejemplo:

    -- Create a temporary table named tablestore_stream for the source data table.
CREATE TEMPORARY TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector' = 'ots',
    'endPoint' = 'https://xxx.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    'instanceName' = 'xxx',
    'tableName' = 'flink_source_table',
    'tunnelName' = 'flink_source_tunnel',
    'accessId' = 'xxxxxxxxxxx',
    'accessKey' = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false' 
);

-- Create a temporary table named print_table for the result table. 
CREATE TEMPORARY TABLE print_table(
   `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
  'connector' = 'print',   -- Use the Print connector.
  'logger' = 'true'        -- Display the computing result in the development console of Realtime Compute for Apache Flink.
);

-- Print the fields of the source table.
INSERT INTO print_table
SELECT `order`,orderid,customerid,customername from tablestore_stream;

    Apéndice 2: Configurar dependencias de VVR

    1. Descargue las dependencias de VVR.

    2. Cargue las dependencias de VVR.

      1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

      2. Busque el espacio de trabajo deseado y haga clic en Console en la columna Actions.

      3. En el panel de navegación izquierdo, haga clic en Artifacts.

      4. En la página Artifacts, haga clic en Upload Artifact y seleccione el paquete JAR de dependencias de VVR.

    3. En el lado derecho del editor SQL del trabajo deseado, haga clic en la pestaña Configurations. En el campo Additional Dependencies, seleccione el paquete JAR de dependencias de VVR.