TSDB for InfluxDB® (en proceso de descontinuación) - Realtime Compute for Apache Flink

Importante

El conector de TSDB for InfluxDB se descontinuará en la versión 11.7. Una vez descontinuado, se eliminará de la consola y dejará de recibir actualizaciones de funciones o mantenimiento. Para consultar el cronograma de descontinuación, véase End of support for the TSDB for InfluxDB connector. Migre sus cargas de trabajo lo antes posible para evitar interrupciones en sus trabajos de producción.

El conector de TSDB for InfluxDB escribe datos de streaming desde una tabla sink de Flink SQL hacia una instancia de TSDB for InfluxDB en Ververica Runtime (VVR). TSDB for InfluxDB es una base de datos de series temporales optimizada para un alto throughput de escritura y consulta; se utiliza comúnmente para el monitoreo de DevOps, métricas de aplicaciones y datos de sensores IoT.

Capacidades del conector

Elemento	Valor
Tipo de tabla	Sink
Modo de ejecución	Streaming
Formato de datos	Point
Tipo de API	SQL
Actualización o eliminación de datos en la tabla sink	No compatible
Métricas	`numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`

Para obtener más información sobre estas métricas, consulte Métricas de monitoreo.

Requisitos previos

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

Una base de datos creada en TSDB for InfluxDB. Consulte Gestionar cuentas de usuario y bases de datosGestionar cuentas de usuario y bases de datosGestionar cuentas de usuario y bases de datosGestionar cuentas de usuario y bases de datos

Limitaciones

El conector de TSDB for InfluxDB solo es compatible con despliegues de Realtime Compute for Apache Flink que usan VVR 2.1.5 o versiones posteriores.

Crear una tabla sink

DDL mínimo

A continuación, el ejemplo muestra las columnas mínimas obligatorias para definir una tabla sink:

CREATE TABLE influxdb_sink (
  `metric`            VARCHAR,
  `timestamp`         BIGINT,
  `tag_value1`        VARCHAR,
  `field_fieldValue1` DOUBLE
) WITH (
  'connector' = 'influxdb',
  'url'       = 'http://service.cn.influxdb.aliyuncs.com:****',
  'database'  = '<yourDatabaseName>',
  'username'  = '<yourDatabaseUserName>',
  'password'  = '<yourDatabasePassword>'
);

Convenciones de nombres de columnas del esquema

Las columnas de la tabla sink deben seguir una convención de nomenclatura fija asignada al modelo de datos de InfluxDB. El orden de las columnas es fijo.

Posición	Nombre de columna	Tipo	Obligatorio	Se asigna a
0	`metric`	VARCHAR	Sí	Nombre de medición de InfluxDB
1	`timestamp`	BIGINT	Sí	Marca de tiempo de InfluxDB; la unidad debe ser milisegundos
2+	`tag_<name>`	VARCHAR	Como mínimo uno	Etiqueta de InfluxDB (metadatos indexados)
3+	`field_<name>`	Cualquier tipo compatible	Como mínimo uno	Campo de InfluxDB (valor de datos)

Para escribir en varias columnas de campo, defínalas con el siguiente patrón:

`field_fieldValue1` DOUBLE,
`field_fieldValue2` INTEGER,
`field_fieldValueN` INTEGER

Nota

Solo se admiten los nombres de columna metric, timestamp, tag_* y field_*. Cualquier otro nombre de columna provoca un error.

Opciones del conector

Parámetro	Obligatorio	Predeterminado	Tipo	Descripción
`connector`	Sí	—	String	Debe ser `influxdb`.
`url`	Sí	—	String	Endpoint de VPC de la instancia de TSDB for InfluxDB. Se admiten tanto HTTP como HTTPS. Ejemplo: `https://localhost:8086` o `http://localhost:3242`.
`database`	Sí	—	String	Nombre de la base de datos. Ejemplo: `db-flink`.
`username`	Sí	—	String	Nombre de usuario de la base de datos. El usuario debe tener permisos de escritura en la base de datos de destino. Consulte Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`password`	Sí	—	String	Contraseña del usuario especificado. Consulte Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`batchSize`	No	`300`	Integer	Número de registros escritos en un solo lote.
`retentionPolicy`	No	`autogen`	String	Directiva de retención de la base de datos de destino. Si no se especifica, se usa la directiva de retención predeterminada de la base de datos (`autogen`). Consulte Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`ignoreErrorData`	No	`false`	Boolean	Manejo de errores de escritura. `true`: ignora los errores de escritura y continúa. `false`: falla el trabajo cuando ocurre un error de escritura.

Asignaciones de tipos de datos

Tipo de InfluxDB	Tipo de Flink
BOOLEAN	BOOLEAN
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DECIMAL	DECIMAL
DOUBLE	DOUBLE
DATE	DATE
TIME	TIME
TIMESTAMP	TIMESTAMP
VARCHAR	VARCHAR

Ejemplo

A continuación, el ejemplo genera datos aleatorios mediante el conector datagen y los escribe en TSDB for InfluxDB.

CREATE TEMPORARY TABLE datagen_source (
  `metric`     VARCHAR,
  `timestamp`  BIGINT,
  `fieldvalue` DOUBLE,
  `tagvalue`   VARCHAR
) WITH (
  'connector'                  = 'datagen',
  'fields.metric.length'       = '3',
  'fields.tagvalue.length'     = '3',
  'fields.timestamp.min'       = '1587539547000',
  'fields.timestamp.max'       = '1619075547000',
  'fields.fieldvalue.min'      = '1',
  'fields.fieldvalue.max'      = '100000',
  'rows-per-second'            = '50'
);

CREATE TEMPORARY TABLE influxdb_sink (
  `metric`            VARCHAR,
  `timestamp`         BIGINT,
  `field_fieldValue1` DOUBLE,
  `tag_value1`        VARCHAR
) WITH (
  'connector'        = 'influxdb',
  'url'              = 'https://***********.influxdata.tsdb.aliyuncs.com:****',
  'database'         = '<yourDatabaseName>',
  'username'         = '<yourDatabaseUserName>',
  'password'         = '<yourDatabasePassword>',
  'batchSize'        = '100',
  'retentionPolicy'  = 'autogen',
  'ignoreErrorData'  = 'false'
);

INSERT INTO influxdb_sink
SELECT
  `metric`,
  `timestamp`,
  `fieldvalue`,
  `tagvalue`
FROM datagen_source;

Realtime Compute for Apache Flink:TSDB for InfluxDB® (en proceso de descontinuación)