Realtime Compute for Apache Flink:Lindorm connector

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

• Un motor de tablas anchas de Lindorm y una tabla de Lindorm. Para obtener más información, consulte Create an instance.

• Conectividad de red entre el clúster de Lindorm y el espacio de trabajo de Flink; por ejemplo, que ambos se encuentren en la misma virtual private cloud (VPC).

Las tablas Lindorm HBase no son compatibles. Solo se admite LindormTable.

Inicio rápido

El siguiente ejemplo genera 10 filas de datos, busca las filas coincidentes en una tabla de dimensiones de Lindorm y escribe el resultado en una tabla sink de Lindorm.

-- Source: generate 10 rows with sequential IDs 0-9
CREATE TEMPORARY TABLE example_source (
  id INT,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '10',
  'fields.id.kind' = 'sequence',
  'fields.id.start' = '0',
  'fields.id.end' = '9'
);

-- Dimension table: look up user details from Lindorm
CREATE TEMPORARY TABLE lindorm_hbase_dim (
  `id`    INT,
  `name`  VARCHAR,
  `birth` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'   = 'lindorm',
  'tablename'   = '${lindorm_dim_table}',
  'seedserver'  = '${lindorm_seed_server}',
  'namespace'   = 'default',
  'username'    = '${lindorm_username}',
  'password'    = '${lindorm_password}'
);

-- Sink table: write enriched records to Lindorm
CREATE TEMPORARY TABLE lindorm_hbase_sink (
  `id`    INT,
  `name`  VARCHAR,
  `birth` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'   = 'lindorm',
  'tablename'   = '${lindorm_sink_table}',
  'seedserver'  = '${lindorm_seed_server}',
  'namespace'   = 'default',
  'username'    = '${lindorm_username}',
  'password'    = '${lindorm_password}'
);

-- Temporal join: enrich source data with the dimension table and write to sink
INSERT INTO lindorm_hbase_sink
SELECT
  s.id,
  d.name,
  d.birth
FROM example_source AS s
JOIN lindorm_hbase_dim AS d FOR SYSTEM_TIME AS OF s.proc_time
  ON s.id = d.id;

Reemplace los marcadores de posición antes de ejecutar:

Sintaxis

CREATE TABLE white_list (
  id     VARCHAR,
  name   VARCHAR,
  age    INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'    = 'lindorm',
  'seedserver'   = '<host:port>',
  'namespace'    = '<yourNamespace>',
  'username'     = '<yourUsername>',
  'password'     = '<yourPassword>',
  'tableName'    = '<yourTableName>',
  'columnFamily' = '<yourColumnFamily>'
);

Opciones del conector

Caché de búsqueda

De forma predeterminada, el conector obtiene cada búsqueda directamente de Lindorm (cache = None). Habilite la caché LRU para reducir la presión de lectura en Lindorm durante uniones de alto throughput.

Con cache = LRU, el conector almacena en memoria las filas accedidas recientemente. Cuando la clave de unión coincide con una fila en caché, el conector devuelve el valor almacenado sin consultar a Lindorm. Una entrada en caché se desaloja cuando:

• La caché alcanza cacheSize filas (las filas más antiguas se desalojan primero).

• La entrada ha permanecido en la caché durante más de cacheTTLMs milisegundos.

Existe un compromiso: un TTL más largo o una caché más grande reduce el tráfico de lectura hacia Lindorm, pero aumenta el riesgo de servir datos obsoletos. Ajuste ambos valores según sus requisitos de throughput y la frecuencia con la que cambian los datos subyacentes.

El conector de Lindorm admite uniones de búsqueda de uno a muchos. Preste especial atención a las estrategias de caché y al throughput cuando las filas de la tabla de dimensiones puedan coincidir con varios eventos del flujo ascendente.

Escrituras idempotentes

Cuando una tabla de Lindorm tiene una clave primaria, todas las escrituras utilizan semántica upsert: cada registro entrante inserta una nueva fila o actualiza la fila existente coincidente. Esto hace que las escrituras sean idempotentes.

Las escrituras idempotentes son importantes para la tolerancia a fallos. Si un trabajo de Flink se reinicia desde un checkpoint, reproduce los mensajes desde el último checkpoint exitoso. Dado que los upserts de Lindorm son idempotentes, los registros reproducidos generan el mismo resultado que las escrituras originales, sin filas duplicadas ni violaciones de restricciones.

Defina una clave primaria en su DDL para aprovechar las escrituras idempotentes.

Tabla dinámica

Utilice la función de tabla dinámica cuando su esquema evolucione en tiempo de ejecución; es decir, cuando las columnas se creen dinámicamente según los valores de los datos en lugar de definirse de forma fija en el DDL. Un caso de uso típico es el seguimiento de métricas horarias por día, donde las horas son nombres de columna y los días son claves primarias:

Reglas de DDL para tablas dinámicas:

• Las primeras N columnas forman la clave primaria.

• Las últimas dos columnas deben ser VARCHAR.

• La penúltima columna (c1) contiene el nombre de la columna dinámica.

• La última columna (c2) contiene el valor de dicha columna.

• No se permiten columnas fuera de la clave primaria distintas de c1 y c2.

CREATE TABLE lindorm_dynamic_output (
  pk1 VARCHAR,
  pk2 VARCHAR,
  pk3 VARCHAR,
  c1  VARCHAR,  -- column name written to Lindorm
  c2  VARCHAR,  -- column value written to Lindorm
  PRIMARY KEY (pk1, pk2, pk3) NOT ENFORCED
) WITH (
  'connector'         = 'lindorm',
  'seedserver'        = '<host:port>',
  'namespace'         = '<yourNamespace>',
  'username'          = '<yourUsername>',
  'password'          = '<yourPassword>',
  'tableName'         = '<yourTableName>',
  'columnFamily'      = '<yourColumnFamily>',
  'dynamicColumnSink' = 'true'
);

Cada vez que se escribe un registro, el conector agrega o actualiza una columna en la fila de Lindorm identificada por <pk1, pk2, pk3>. Las demás columnas de esa fila permanecen sin cambios.

Mapeo de tipos de datos

Todos los datos de Lindorm se almacenan en formato binario. La siguiente tabla muestra cómo convierte el conector entre los tipos de Flink SQL y las representaciones binarias de Lindorm.

Los métodos de Bytes se encuentran en la clase com.alibaba.lindorm.client.core.utils.Bytes. Los métodos de StringData (filas CHAR/VARCHAR) se encuentran en la clase org.apache.flink.table.data.StringData.

Métricas

A continuación se presentan las métricas de sink disponibles. Para más detalles, consulte Metrics.

Preguntas frecuentes

Lindorm connection errors and solutions