ApsaraDB for HBase connector - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Contexto

ApsaraDB for HBase es un servicio NoSQL inteligente basado en la nube y rentable que ofrece alta escalabilidad y compatibilidad con HBase de código abierto. ApsaraDB for HBase proporciona beneficios como bajos costos de almacenamiento, alto throughput y escalabilidad, además de procesamiento inteligente de datos. ApsaraDB for HBase admite servicios centrales de Alibaba, como las recomendaciones de Taobao, el control de riesgos de Ant Credit Pay, publicidad, paneles de datos, seguimiento logístico de Cainiao, registros de transacciones de Alipay y mensajes de Taobao Mobile. ApsaraDB for HBase es un servicio totalmente administrado que ofrece capacidades empresariales, como el procesamiento de petabytes de datos, alta concurrencia, escalado rápido en segundos, baja latencia de respuesta en milisegundos, alta disponibilidad entre centros de datos y distribución global.

La siguiente tabla describe las funciones compatibles con el conector de ApsaraDB for HBase.

Elemento	Descripción
Tipo de tabla	Tabla de dimensiones y tabla sink
Modo de ejecución	Modo streaming
Formato de datos	N/A
Métrica	Métricas Source Ninguna Tablas de dimensiones Ninguna Sink numBytesOut, numBytesOutPerSecond, numRecordsOut, numRecordsOutPerSecond y currentSendTime Nota Para obtener más información sobre las métricas, consulte Metrics.
Tipo de API	API SQL
Actualización o eliminación de datos en una tabla sink	Compatible

Requisitos previos

Debe adquirir un clúster de ApsaraDB for HBase y crear una tabla de ApsaraDB for HBase. Para obtener más información sobre cómo adquirir un clúster de ApsaraDB for HBase, consulte Purchase a cluster.
Configure una lista de permitidos para el clúster de ApsaraDB for HBase. Para obtener más información, consulte Configure a whitelist.

Notas de uso

Antes de usar el conector de ApsaraDB for HBase, confirme el tipo de su instancia de base de datos y asegúrese de que el tipo de conector seleccionado sea el correcto. Si usa el conector de forma inadecuada, pueden surgir problemas inesperados.

El conector de ApsaraDB for HBase descrito en este tema está diseñado para instancias de ApsaraDB for HBase.
Las instancias de Lindorm son compatibles con Apache HBase. Use el conector de Lindorm para las instancias de Lindorm. Para obtener más información, consulte Lindorm.
Si usa el conector de ApsaraDB for HBase para conectar Realtime Compute for Apache Flink a una base de datos HBase de código abierto, no se puede garantizar la validez de los datos.

Sintaxis

CREATE TABLE hbase_table(
  rowkey INT,
  family1 ROW<q1 INT>,
  family2 ROW<q2 STRING, q3 BIGINT>,
  family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>
) WITH (
  'connector'='cloudhbase',
  'table-name'='<yourTableName>',
  'zookeeper.quorum'='<yourZookeeperQuorum>'
);

Las familias de columnas de una tabla de ApsaraDB for HBase deben declararse con el tipo ROW. El nombre de cada familia de columnas corresponde al nombre de campo de una fila. En la sintaxis DDL se declaran las siguientes familias de columnas: family1, family2 y family3.
Una columna dentro de una familia de columnas corresponde a un campo en una fila. El nombre de la columna es el nombre del campo. En la sintaxis DDL, las columnas q2 y q3 se declaran en la familia de columnas family2.
Además de los campos de tipo ROW, solo puede existir un campo de tipo atómico, como STRING o BIGINT, en una tabla de ApsaraDB for HBase. El campo de tipo atómico se considera la clave de fila de la tabla, como rowkey en la sentencia DDL.
La clave de fila de una tabla de ApsaraDB for HBase debe definirse como la clave primaria de la tabla sink. Si no se define ninguna clave primaria, la clave de fila se usa como tal.
Solo es necesario declarar en la tabla sink las familias de columnas y las columnas requeridas de la tabla de ApsaraDB for HBase.

Opciones del conector

Generales

Opción	Descripción	Tipo de dato	Obligatorio	Valor predeterminado	Observaciones
connector	El tipo de la tabla.	String	Sí	Sin valor predeterminado	Establezca el valor en `cloudhbase`.
table-name	El nombre de la tabla de ApsaraDB for HBase.	String	Sí	Sin valor predeterminado	N/A.
zookeeper.znode.quorum	La URL usada para acceder al servicio ZooKeeper de ApsaraDB for HBase.	String	Sí	Sin valor predeterminado	N/A.
zookeeper.znode.parent	El directorio raíz de ApsaraDB for HBase en el servicio ZooKeeper.	String	No	`/hbase`	Este parámetro solo surte efecto en ApsaraDB for HBase Standard Edition.
userName	El nombre de usuario usado para acceder a la base de datos.	String	No	Sin valor predeterminado	Este parámetro solo surte efecto en ApsaraDB for HBase Performance-enhanced Edition.
password	La contraseña usada para acceder a la base de datos.	String	No	Sin valor predeterminado	Este parámetro solo surte efecto en ApsaraDB for HBase Performance-enhanced Edition.
haclient.cluster.id	El ID del clúster de ApsaraDB for HBase en modo de alta disponibilidad (HA).	String	No	Sin valor predeterminado	Este parámetro solo es obligatorio cuando se accede a clústeres de recuperación ante desastres por zona. Este parámetro solo surte efecto en ApsaraDB for HBase Performance-enhanced Edition.
retires.number	El número de reintentos permitidos para que el cliente de ApsaraDB for HBase se conecte a la base de datos de ApsaraDB for HBase.	Integer	No	31	N/A.
null-string-literal	Si el tipo de dato de un campo de ApsaraDB for HBase es STRING y los datos del campo en Realtime Compute for Apache Flink son nulos, se asigna `null-string-literal` al campo de ApsaraDB for HBase y se escribe en la base de datos de ApsaraDB for HBase.	String	No	null	N/A.

Específicas de sink

Opción	Descripción	Tipo de dato	Obligatorio	Valor predeterminado	Observaciones
sink.buffer-flush.max-size	El tamaño de los datos en bytes almacenados en caché en la memoria antes de escribirlos en la base de datos de ApsaraDB for HBase. Un valor mayor mejora el rendimiento de escritura de ApsaraDB for HBase, pero aumenta la latencia de escritura y consume más memoria.	String	No	2 MB	Unidad: B, KB, MB o GB. La unidad no distingue entre mayúsculas y minúsculas. Si establece este parámetro en 0, no se almacenan datos en caché.
sink.buffer-flush.max-rows	El número de registros de datos almacenados en caché en la memoria antes de escribirlos en la base de datos de ApsaraDB for HBase. Un valor mayor mejora el rendimiento de escritura de ApsaraDB for HBase, pero aumenta la latencia de escritura y consume más memoria.	Integer	No	1000	Si establece este parámetro en 0, no se almacenan datos en caché.
sink.buffer-flush.interval	El intervalo en el que los datos en caché se escriben en la base de datos de ApsaraDB for HBase. Este parámetro controla la latencia de escritura de datos en la base de datos de ApsaraDB for HBase.	Duration	No	1 s	Unidad: ms, s, min, h o d. Si establece este parámetro en 0, se deshabilita la escritura periódica de datos.
dynamic.table	Especifica si se debe usar una tabla de ApsaraDB for HBase que admita columnas dinámicas.	Boolean	No	false	Valores válidos: true false
sink.ignore-delete	Especifica si se deben ignorar los mensajes de retractación.	Boolean	No	false	Si un flujo contiene eventos `DELETE` o `UPDATE_BEFORE`, y varias tareas sink actualizan simultáneamente diferentes campos de una tabla, puede producirse inconsistencia en los datos. Por ejemplo, después de eliminar un registro, otra tarea actualiza algunos campos. Los campos no actualizados pasarán a ser nulos o se rellenarán con valores predeterminados, lo que provocará errores en los datos. Para evitar este problema, establezca `sink.ignore-delete` en `true` para ignorar los eventos `DELETE` y `UPDATE_BEFORE` provenientes del flujo de entrada. Nota `UPDATE_BEFORE` forma parte del mecanismo de retractación de Flink y se usa para retractar el valor anterior durante una operación de actualización. Si `ignoreDelete` se establece en `true`, se omiten todos los eventos `DELETE` y `UPDATE_BEFORE`. Solo se procesan los registros `INSERT` y `UPDATE_AFTER`.
sink.sync-write	Especifica si se deben escribir datos en ApsaraDB for HBase en modo síncrono.	Boolean	No	true	Valores válidos: true: Los datos se escriben en modo síncrono. En este modo, los datos se escriben en secuencia, pero el rendimiento de escritura se ve afectado. false: Los datos se escriben en modo asíncrono. En este modo, es posible que los datos no se escriban en secuencia, pero el rendimiento de escritura mejora.
sink.buffer-flush.batch-rows	El número de registros de datos que se almacenan en caché en la memoria cuando los datos se escriben en ApsaraDB for HBase en modo síncrono. Un valor mayor mejora el rendimiento de escritura de ApsaraDB for HBase, pero aumenta la latencia de escritura y el uso de memoria.	Integer	No	100	Este parámetro solo surte efecto cuando el parámetro sink.sync-write se establece en true.
sink.ignore-null	Especifica si se deben ignorar los valores nulos.	Boolean	No	false	Nota Si establece este parámetro en true, el parámetro `null-string-literal` no surte efecto. Solo Realtime Compute for Apache Flink con VVR 8.0.9 o posterior admite este parámetro.

Opciones específicas de tablas de dimensiones (relacionadas con caché)

Opción	Descripción	Tipo de dato	Obligatorio	Valor predeterminado	Observaciones
cache	La directiva de caché.	String	No	ALL	Valores válidos: None: No se almacenan datos en caché. LRU: Solo se almacenan en caché datos específicos de la tabla de dimensiones. Cada vez que el sistema recibe un registro de datos, busca en la caché. Si el sistema no encuentra el registro en la caché, busca el registro de datos en la tabla de dimensiones física. Nota Si usa esta directiva de caché, configure los parámetros cacheSize y cacheTTLMs. ALL: Se almacenan en caché todos los datos de la tabla de dimensiones. Este es el valor predeterminado. Antes de que se ejecute un trabajo, el sistema carga todos los datos de la tabla de dimensiones en la caché. De este modo, todas las consultas posteriores en la tabla de dimensiones se buscan en la caché. Si el sistema no encuentra el registro de datos en la caché, significa que la clave de unión no existe. El sistema recarga todos los datos en la caché después de que expiren las entradas de la caché. Nota Si la cantidad de datos en una tabla remota es pequeña y existe un gran número de claves faltantes, se recomienda establecer este parámetro en ALL. La tabla origen y la tabla de dimensiones no se pueden asociar mediante la cláusula ON. Si usa esta directiva de caché, configure los parámetros cacheTTLMs y cacheReloadTimeBlackList. Si todos los datos de la tabla de dimensiones se cargan en la caché, la velocidad de inicio del despliegue puede disminuir. Configure la directiva de caché de manera flexible según sus requisitos empresariales. Si establece el parámetro cache en ALL, aumente la memoria del nodo para la unión de tablas, ya que el sistema carga datos de forma asíncrona desde la tabla de dimensiones. El tamaño de memoria adicional equivale al doble del tamaño de la tabla remota.
cacheSize	El número máximo de filas de datos que se pueden almacenar en caché.	Long	No	10000	Configure este parámetro cuando establezca el parámetro cache en LRU.
cacheTTLMs	El periodo de tiempo de espera de la caché. Unidad: milisegundos.	Long	No	Sin valor predeterminado	La configuración del parámetro cacheTTLMs varía según el parámetro cache. Si establece el parámetro cache en None, el parámetro cacheTTLMs puede dejarse vacío. Esto indica que las entradas de la caché no caducan. Si establece el parámetro cache en LRU, el parámetro cacheTTLMs especifica el periodo de tiempo de espera de la caché. De forma predeterminada, las entradas de la caché no caducan. Si establece el parámetro cache en ALL, el parámetro cacheTTLMs especifica el intervalo en el que el sistema recarga la caché. De forma predeterminada, la caché no se recarga.
cacheEmpty	Especifica si se deben almacenar en caché los resultados vacíos.	Boolean	No	true	N/A.
cacheReloadTimeBlackList	Los periodos de tiempo durante los cuales no se actualiza la caché. Este parámetro surte efecto cuando el parámetro cache se establece en ALL. La caché no se actualiza durante los periodos de tiempo que especifique para este parámetro. Este parámetro es adecuado para eventos promocionales en línea a gran escala, como el Double 11.	String	No	Sin valor predeterminado	El siguiente ejemplo muestra el formato de los valores: 2017-10-24 14:00 -> 2017-10-24 15:00, 2017-11-10 23:30 -> 2017-11-11 08:00. Use delimitadores según las siguientes reglas: Separe varios periodos de tiempo con comas (,). Separe la hora de inicio y la hora de finalización de cada periodo de tiempo con una flecha (->) que es una combinación de un guion (-) y un signo de mayor que (>).
cacheScanLimit	El número de filas que el servidor de llamada a procedimiento remoto (RPC) devuelve a un cliente cuando el servidor lee todos los datos de una tabla de dimensiones de ApsaraDB for HBase.	Integer	No	100	Este parámetro solo está disponible cuando establece el parámetro cache en ALL.

Asignaciones de tipos de datos

Un valor de un tipo de dato de Realtime Compute for Apache Flink se convierte en una matriz de bytes mediante org.apache.hadoop.hbase.util.Bytes en una tabla de ApsaraDB for HBase. El proceso de decodificación varía según los siguientes escenarios:

Si el tipo de dato de Realtime Compute for Apache Flink no es STRING y un valor en la tabla de ApsaraDB for HBase es una matriz de bytes vacía, el valor se decodifica como nulo.
Si el tipo de dato de Realtime Compute for Apache Flink es STRING y un valor en la tabla de dimensiones de ApsaraDB for HBase es la matriz de bytes especificada por null-string-literal, el valor se decodifica como nulo.

Tipo de Flink SQL	Función para convertir un valor en bytes para ApsaraDB for HBase	Función para leer bytes desde ApsaraDB for HBase
CHAR	byte[] toBytes(String s)	String toString(byte[] b)
VARCHAR
STRING
BOOLEAN	byte[] toBytes(boolean b)	boolean toBoolean(byte[] b)
BINARY	byte[]	byte[]
VARBINARY	byte[]	byte[]
DECIMAL	byte[] toBytes(BigDecimal v)	BigDecimal toBigDecimal(byte[] b)
TINYINT	new byte[] { val }	bytes[0]
SMALLINT	byte[] toBytes(short val)	short toShort(byte[] bytes)
INT	byte[] toBytes(int val)	int toInt(byte[] bytes)
BIGINT	byte[] toBytes(long val)	long toLong(byte[] bytes)
FLOAT	byte[] toBytes(float val)	float toFloat(byte[] bytes)
DOUBLE	byte[] toBytes(double val)	double toDouble(byte[] bytes)
DATE	Convierte una fecha en un valor INT que representa el número de días desde el 1 de enero de 1970 y luego en una matriz de bytes mediante `byte[] toBytes(int val)`.	Convierte una matriz de bytes de la base de datos de ApsaraDB for HBase en el tipo de dato INT mediante `int toInt(byte[] bytes)`. El valor del tipo de dato INT representa el número de días desde el 1 de enero de 1970.
TIME	Convierte una hora en un valor INT que representa el número de milisegundos desde las 00:00:00 y luego en una matriz de bytes mediante `byte[] toBytes(int val)`.	Convierte una matriz de bytes de la base de datos de ApsaraDB for HBase en el tipo de dato INT mediante `int toInt(byte[] bytes)`. El valor del tipo de dato INT representa el número de milisegundos desde las 00:00:00.
TIMESTAMP	Convierte una marca de tiempo en un valor LONG que representa el número de milisegundos desde las 00:00:00 del 1 de enero de 1970 y luego en una matriz de bytes mediante `byte[] toBytes(long val)`.	Convierte una matriz de bytes de la base de datos de ApsaraDB for HBase en el tipo de dato LONG mediante `long toLong(byte[] bytes)`. El valor del tipo de dato LONG representa el número de milisegundos desde las 00:00:00 del 1 de enero de 1970.

Código de ejemplo

Código de ejemplo para una tabla de dimensiones

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  `proc_time` AS PROCTIME()
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hbase_dim (
  rowkey INT,
  family1 ROW<col1 INT>,
  family2 ROW<col1 STRING, col2 BIGINT>,
  family3 ROW<col1 DOUBLE, col2 BOOLEAN, col3 STRING>
) WITH (
  'connector' = 'cloudhbase',
  'table-name' = '<yourTableName>',
  'zookeeper.quorum' = '<yourZookeeperQuorum>'
);

CREATE TEMPORARY TABLE blackhole_sink(
  a INT,
  f1c1 INT,
  f3c3 STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
     SELECT a, family1.col1 as f1c1,  family3.col3 as f3c3 FROM datagen_source
JOIN hbase_dim FOR SYSTEM_TIME AS OF datagen_source.`proc_time` as h ON datagen_source.a = h.rowkey;

Código de ejemplo para una tabla sink

CREATE TEMPORARY TABLE datagen_source (
  rowkey INT,
  f1q1 INT,
  f2q1 STRING,
  f2q2 BIGINT,
  f3q1 DOUBLE,
  f3q2 BOOLEAN,
  f3q3 STRING
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hbase_sink (
  rowkey INT,
  family1 ROW<q1 INT>,
  family2 ROW<q1 STRING, q2 BIGINT>,
  family3 ROW<q1 DOUBLE, q2 BOOLEAN, q3 STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector'='cloudhbase',
  'table-name'='<yourTableName>',
  'zookeeper.quorum'='<yourZookeeperQuorum>'
);
 
INSERT INTO hbase_sink
SELECT rowkey, ROW(f1q1), ROW(f2q1, f2q2), ROW(f3q1, f3q2, f3q3) FROM datagen_source;

Código de ejemplo para una tabla sink que admite columnas dinámicas
```
CREATE TEMPORARY TABLE datagen_source (
  id INT,
  f1hour STRING,
  f1deal BIGINT,
  f2day STRING,
  f2deal BIGINT
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hbase_sink (
  rowkey INT,
  f1 ROW<`hour` STRING, deal BIGINT>,
  f2 ROW<`day` STRING, deal BIGINT>
) WITH (
  'connector'='cloudhbase',
  'table-name'='<yourTableName>',
  'zookeeper.quorum'='<yourZookeeperQuorum>',
  'dynamic.table'='true'
);

INSERT INTO hbase_sink
SELECT id, ROW(f1hour, f1deal), ROW(f2day, f2deal) FROM datagen_source;
```
- Si dynamic.table se establece en true, se usa una tabla de ApsaraDB for HBase que admite columnas dinámicas.
- Declare dos campos en las filas que corresponden a cada familia de columnas. El valor del primer campo indica la columna dinámica, y el valor del segundo campo indica el valor de la columna dinámica.
- Por ejemplo, la tabla datagen_source contiene una fila de datos. La fila de datos indica que el ID del producto es 1, el monto de la transacción del producto entre las 10:00 y las 11:00 es 100, y el monto de la transacción del producto el 26 de julio de 2020 es 10000. En este caso, se inserta una fila cuya rowkey es 1 en la tabla de ApsaraDB for HBase. f1:10 es 100, y f2:2020-7-26 es 10000.