Crie uma tabela sink do InfluxDB e código de exemplo - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Importante

O conector do TSDB for InfluxDB será descontinuado na versão 11.7. Após a descontinuação, ele será removido do console e não receberá mais atualizações de recursos ou manutenção. Para consultar o cronograma de descontinuação, veja Fim do suporte para o conector do TSDB for InfluxDB. Migre suas cargas de trabalho o quanto antes para evitar interrupções em seus jobs de produção.

O conector do TSDB for InfluxDB grava dados de streaming de uma tabela sink do Flink SQL em uma instância do TSDB for InfluxDB no Ververica Runtime (VVR). O TSDB for InfluxDB é um banco de dados de séries temporais otimizado para alta taxa de gravação e consulta, comumente utilizado para monitoramento de DevOps, métricas de aplicações e dados de sensores IoT.

Capacidades do conector

Item	Valor
Tipo de tabela	Sink
Modo de execução	Streaming
Formato de dados	Point
Tipo de API	SQL
Atualização ou exclusão de dados na tabela sink	Não suportado
Métricas	`numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`

Para obter detalhes sobre essas métricas, consulte Métricas de monitoramento.

Pré-requisitos

Antes de começar, certifique-se de ter:

Um banco de dados criado no TSDB for InfluxDB. Consulte Gerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dados

Limitações

O conector do TSDB for InfluxDB é compatível apenas com implantações do Realtime Compute for Apache Flink que utilizam VVR 2.1.5 ou superior.

Crie uma tabela sink

DDL mínimo

O exemplo a seguir apresenta as colunas mínimas necessárias para definir uma tabela 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>'
);

Convenções de nomenclatura das colunas do schema

As colunas da tabela sink devem seguir uma convenção de nomenclatura fixa que mapeia para o modelo de dados do InfluxDB. A ordem das colunas também é fixa.

Posição	Nome da coluna	Tipo	Obrigatório	Mapeia para
0	`metric`	VARCHAR	Sim	Nome da measurement do InfluxDB
1	`timestamp`	BIGINT	Sim	Timestamp do InfluxDB; a unidade deve ser milissegundos
2+	`tag_<name>`	VARCHAR	Pelo menos uma	Tag do InfluxDB (metadados indexados)
3+	`field_<name>`	Qualquer tipo suportado	Pelo menos um	Field do InfluxDB (valor de dado)

Para gravar em múltiplas colunas de field, defina-as utilizando o seguinte padrão:

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

Nota

Apenas os nomes de coluna metric, timestamp, tag_* e field_* são suportados. Qualquer outro nome de coluna causará um erro.

Opções do conector

Parâmetro	Obrigatório	Padrão	Tipo	Descrição
`connector`	Sim	—	String	Deve ser `influxdb`.
`url`	Sim	—	String	Endpoint de VPC da instância do TSDB for InfluxDB. Tanto HTTP quanto HTTPS são suportados. Exemplo: `https://localhost:8086` ou `http://localhost:3242`.
`database`	Sim	—	String	Nome do banco de dados. Exemplo: `db-flink`.
`username`	Sim	—	String	Nome de usuário do banco de dados. O usuário deve ter permissões de gravação no banco de dados de destino. Consulte Gerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dados.
`password`	Sim	—	String	Senha do usuário especificado. Consulte Gerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dados.
`batchSize`	Não	`300`	Integer	Número de registros a serem gravados em um único lote.
`retentionPolicy`	Não	`autogen`	String	Política de retenção do banco de dados de destino. Se não for especificada, a política de retenção padrão do banco de dados (`autogen`) será utilizada. Consulte Gerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dadosGerencie contas de usuário e bancos de dados.
`ignoreErrorData`	Não	`false`	Boolean	Define como tratar erros de gravação. `true`: ignora erros de gravação e continua. `false`: falha o job quando ocorre um erro de gravação.

Mapeamentos de tipos de dados

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

Exemplo

O exemplo abaixo gera dados aleatórios usando o conector datagen e os grava no 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® (descontinuação)