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

El conector de DataHub permite leer datos de streaming desde Alibaba Cloud DataHub hacia trabajos de Flink y escribir los resultados procesados de vuelta en temas de DataHub. Es compatible tanto con Flink SQL como con la API DataStream.

Nota

DataHub es compatible con el protocolo Kafka. Para conectar Flink a DataHub mediante el protocolo Kafka, utilice el conector estándar de Kafka, no el conector Upsert Kafka. Para más detalles, consulte

Compatibilidad con Kafka

Capacidades

Elemento	Descripción
Tipo compatible	Origen y destino
Modo de ejecución	Streaming y por lotes
Formato de datos	N/A
Métricas	N/A
Tipo de API	DataStream y SQL
Compatibilidad con actualización o eliminación de datos en el destino	No compatible. El destino escribe únicamente filas de inserción en el tema de destino.

Requisitos previos

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

Un proyecto y un tema de DataHub. Consulte Get started with DataHub.
Una suscripción de DataHub (obligatoria para el origen). Consulte Create a subscription.
Un AccessKey ID y un AccessKey secret de Alibaba Cloud. Consulte Console operations.

Límites

No se recomienda leer una tabla de origen de DataHub con un trabajo por lotes. En modo por lotes, la tabla de origen de DataHub no alcanza el estado finalizado, por lo que el trabajo permanece en espera en lugar de completarse.

Sintaxis

CREATE TEMPORARY TABLE datahub_input (
  `time` BIGINT,
  `sequence`  STRING METADATA VIRTUAL,
  `shard-id` BIGINT METADATA VIRTUAL,
  `system-time` TIMESTAMP METADATA VIRTUAL
) WITH (
  'connector' = 'datahub',
  'subId' = '<yourSubId>',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

Opciones del conector

Generales

Opción	Tipo	Obligatorio	Predeterminado	Descripción
connector	String	Sí	(ninguno)	El tipo de conector. Establezca este valor en `datahub`.
endPoint	String	Sí	(ninguno)	El endpoint del proyecto de DataHub. El valor varía según la región. Consulte Endpoints.
project	String	Sí	(ninguno)	El nombre del proyecto de DataHub.
topic	String	Sí	(ninguno)	El nombre del tema de DataHub. Para temas BLOB (datos sin tipo ni estructura), la tabla de Flink debe contener exactamente una columna VARBINARY.
accessId	String	Sí	(ninguno)	El AccessKey ID de su cuenta de Alibaba Cloud. Guárdelo como variable en lugar de codificarlo directamente. Consulte Gestionar variables.
accessKey	String	Sí	(ninguno)	El AccessKey secret de su cuenta de Alibaba Cloud.
retryTimeout	Integer	No	1800000	El tiempo máximo de espera para un intento de reintento, en milisegundos.
retryInterval	Integer	No	1000	El intervalo entre intentos de reintento, en milisegundos.
CompressType	String	No	lz4	El algoritmo de compresión para lecturas y escrituras. Valores válidos: `lz4`, `deflate`, `""` (deshabilitado). Requiere VVR 6.0.5 o posterior.

Específicas del origen

Opción	Tipo	Obligatorio	Predeterminado	Descripción
subId	String	Sí	(ninguno)	El ID de suscripción de DataHub.
maxFetchSize	Integer	No	50	El número de registros obtenidos por solicitud. Aumente este valor para mejorar el throughput de lectura.
maxBufferSize	Integer	No	50	El número máximo de registros almacenados en caché desde lecturas asíncronas. Aumente este valor para mejorar el throughput de lectura.
fetchLatestDelay	Integer	No	500	La duración de espera en milisegundos cuando no hay datos disponibles. Reduzca este valor para disminuir la latencia de lectura en temas con poco tráfico.
lengthCheck	String	No	NONE	La regla para manejar filas donde el recuento de campos analizados no coincide con el recuento de columnas definidas. Valores válidos: `NONE`, `SKIP`, `EXCEPTION`, `PAD`. Consulte Reglas de validación de recuento de campos.
columnErrorDebug	Boolean	No	false	Especifica si se habilita el registro de depuración para errores de análisis de campos. Establezca este valor en `true` para imprimir registros de excepciones de análisis.
startTime	String	No	(ninguno)	La marca de tiempo desde la cual comenzar a consumir. Formato: `yyyy-MM-dd hh:mm:ss`.
endTime	String	No	(ninguno)	La marca de tiempo en la cual detener el consumo. Formato: `yyyy-MM-dd hh:mm:ss`.
startTimeMs	Long	No	-1	La marca de tiempo desde la cual comenzar a consumir, en milisegundos. Tiene prioridad sobre `startTime`. Consulte Posición inicial de consumo.

Posición inicial de consumo

La opción startTimeMs controla dónde comienza a leer el origen:

-1 (predeterminado): Comienza desde el desplazamiento más reciente del tema. Si no existe ningún desplazamiento, retrocede al desplazamiento más antiguo.
Una marca de tiempo específica: Comienza desde el primer registro en la marca de tiempo especificada o posterior a ella.

Importante

El valor predeterminado de -1 puede provocar pérdida de datos. Si el trabajo falla antes de su primer punto de control, es posible que el desplazamiento más reciente del tema haya avanzado, y se omitirán los registros escritos durante ese intervalo. Establezca startTimeMs explícitamente en una marca de tiempo específica para controlar la posición inicial.

Reglas de validación de recuento de campos

La opción lengthCheck determina qué sucede cuando el número de campos analizados en una fila no coincide con el número de columnas definidas:

Valor	Comportamiento
`NONE` (predeterminado)	Si los campos analizados > columnas definidas: lee de izquierda a derecha hasta alcanzar el recuento definido. Si los campos analizados < columnas definidas: omite la fila.
`SKIP`	Omite las filas donde el recuento de campos analizados difiere del recuento de columnas definidas.
`EXCEPTION`	Lanza una excepción cuando el recuento de campos analizados difiere del recuento de columnas definidas.
`PAD`	Lee de izquierda a derecha. Si los campos analizados > columnas definidas: lee de izquierda a derecha hasta alcanzar el recuento definido. Si los campos analizados < columnas definidas: rellena los campos faltantes con `null`.

Específicas del destino

Opción	Tipo	Obligatorio	Predeterminado	Descripción
batchCount	Integer	No	500	El número máximo de filas por lote de escritura.
batchSize	Integer	No	512000	El tamaño máximo de un lote de escritura, en bytes.
flushInterval	Integer	No	5000	El intervalo de vaciado, en milisegundos.
hashFields	String	No	null	Una lista separada por comas de nombres de columnas utilizada para enrutar filas a shards. Las filas con los mismos valores en estas columnas se escriben en el mismo shard. El valor predeterminado (`null`) utiliza escrituras aleatorias. Ejemplo: `hashFields=a,b`.
timeZone	String	No	(ninguno)	La zona horaria utilizada al convertir campos TIMESTAMP.
schemaVersion	Integer	No	-1	La versión del esquema en el registro de esquemas registrado.

Comportamiento de vaciado de lotes de escritura

Un lote de escritura se vacía a DataHub cuando se cumple primero cualquiera de las siguientes condiciones:

El número de filas almacenadas en búfer alcanza batchCount.
El tamaño total de los datos almacenados en búfer alcanza batchSize.
El tiempo transcurrido desde el último vaciado supera flushInterval.

Aumentar batchCount, batchSize o flushInterval mejora el throughput de escritura a costa de una mayor latencia.

Asignaciones de tipos de datos

Tipo de Flink	Tipo de DataHub
TINYINT	TINYINT
BOOLEAN	BOOLEAN
INTEGER	INTEGER
BIGINT	BIGINT
BIGINT	TIMESTAMP
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL	DECIMAL
VARCHAR	STRING
SMALLINT	SMALLINT
VARBINARY	BLOB

Metadatos

Los campos de metadatos son de solo lectura (R). Declárelos como METADATA VIRTUAL en la definición de la tabla de origen para incluirlos en las consultas sin escribirlos de vuelta en DataHub.

Nota

Los campos de metadatos están disponibles únicamente al usar VVR 3.0.1 o posterior.

Clave	Tipo de dato	Descripción	L/E
shard-id	BIGINT METADATA VIRTUAL	El ID de shard del registro.	L
sequence	STRING METADATA VIRTUAL	El número de secuencia del registro dentro del shard.	L
system-time	TIMESTAMP METADATA VIRTUAL	La hora en la que DataHub recibió el registro.	L

Ejemplos

Origen

El siguiente ejemplo lee datos de un tema de DataHub y los imprime en la consola.

CREATE TEMPORARY TABLE datahub_input (
  `time` BIGINT,
  `sequence`  STRING METADATA VIRTUAL,
  `shard-id` BIGINT METADATA VIRTUAL,
  `system-time` TIMESTAMP METADATA VIRTUAL
) WITH (
  'connector' = 'datahub',
  'subId' = '<yourSubId>',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

CREATE TEMPORARY TABLE test_out (
  `time` BIGINT,
  `sequence`  STRING,
  `shard-id` BIGINT,
  `system-time` TIMESTAMP
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

INSERT INTO test_out
SELECT
  `time`,
  `sequence`,
  `shard-id`,
  `system-time`
FROM datahub_input;

Destino

El siguiente ejemplo lee desde un tema de DataHub, convierte el campo name a minúsculas y escribe los resultados en otro tema de DataHub.

CREATE TEMPORARY TABLE datahub_source (
  name VARCHAR
) WITH (
  'connector' = 'datahub',
  'endPoint' = '<endPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'subId' = '<yourSubId>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'startTime' = '2018-06-01 00:00:00'
);

CREATE TEMPORARY TABLE datahub_sink (
  name VARCHAR
) WITH (
  'connector' = 'datahub',
  'endPoint' = '<endPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'batchSize' = '512000',
  'batchCount' = '500'
);

INSERT INTO datahub_sink
SELECT
  LOWER(name)
FROM datahub_source;

API DataStream

Importante

Para utilizar la API DataStream con DataHub, configure un conector DataStream para Realtime Compute for Apache Flink. Consulte Settings of DataStream connectors.

Lectura desde DataHub

VVR proporciona la clase DatahubSourceFunction, que implementa la interfaz SourceFunction de Flink.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// Configure the DataHub source
DatahubSourceFunction datahubSource =
    new DatahubSourceFunction(
        <yourEndPoint>,
        <yourProjectName>,
        <yourTopicName>,
        <yourSubId>,
        <yourAccessId>,
        <yourAccessKey>,
        "public",
        <yourStartTime>,
        <yourEndTime>
    );
datahubSource.setRequestTimeout(30 * 1000);
datahubSource.enableExitAfterReadFinished();

env.addSource(datahubSource)
    .map((MapFunction<RecordEntry, Tuple2<String, Long>>) this::getStringLongTuple2)
    .print();
env.execute();

private Tuple2<String, Long> getStringLongTuple2(RecordEntry recordEntry) {
    Tuple2<String, Long> tuple2 = new Tuple2<>();
    TupleRecordData recordData = (TupleRecordData) (recordEntry.getRecordData());
    tuple2.f0 = (String) recordData.getField(0);
    tuple2.f1 = (Long) recordData.getField(1);
    return tuple2;
}

Escritura en DataHub

VVR proporciona la clase OutputFormatSinkFunction, que implementa la interfaz DatahubSinkFunction.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Configure the DataHub sink
env.generateSequence(0, 100)
    .map((MapFunction<Long, RecordEntry>) aLong -> getRecordEntry(aLong, "default:"))
    .addSink(
        new DatahubSinkFunction<>(
            <yourEndPoint>,
            <yourProjectName>,
            <yourTopicName>,
            <yourSubId>,
            <yourAccessId>,
            <yourAccessKey>,
            "public",
            <schemaVersion> // If schema registry is enabled, you must specify the schema version. Otherwise, set this to 0.
        )
    );
env.execute();

private RecordEntry getRecordEntry(Long message, String s) {
    RecordSchema recordSchema = new RecordSchema();
    recordSchema.addField(new Field("f1", FieldType.STRING));
    recordSchema.addField(new Field("f2", FieldType.BIGINT));
    recordSchema.addField(new Field("f3", FieldType.DOUBLE));
    recordSchema.addField(new Field("f4", FieldType.BOOLEAN));
    recordSchema.addField(new Field("f5", FieldType.TIMESTAMP));
    recordSchema.addField(new Field("f6", FieldType.DECIMAL));
    RecordEntry recordEntry = new RecordEntry();
    TupleRecordData recordData = new TupleRecordData(recordSchema);
    recordData.setField(0, s + message);
    recordData.setField(1, message);
    recordEntry.setRecordData(recordData);
    return recordEntry;
}

Dependencia de Maven

Agregue el conector DataStream de DataHub a su proyecto. Todas las versiones disponibles se encuentran listadas en el repositorio central de Maven.

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-datahub</artifactId>
    <version>${vvr-version}</version>
</dependency>

Pasos siguientes

Para obtener una lista completa de conectores compatibles con Realtime Compute for Apache Flink, consulte Conectores compatibles.
Para conectarse a DataHub mediante el conector de Kafka en su lugar, consulte Message Queue for Apache Kafka.
How do I resume a deployment that fails after a DataHub topic is split or scaled in?
Can I delete a DataHub topic that is being consumed?