Criar 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 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 nem manutenção. Para o cronograma de descontinuação, consulte Fim do suporte para o conector TSDB for InfluxDB. Migre suas cargas de trabalho o mais rápido possível para evitar interrupções nos seus jobs de produção.

O conector TSDB for InfluxDB grava dados em streaming de uma tabela sink do Flink SQL em uma instância do TSDB for InfluxDB do 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 usado para monitoramento de DevOps, métricas de aplicativos e dados de sensores de 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 que você tem:

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

Limitações

O conector TSDB for InfluxDB é suportado apenas por deployments do Realtime Compute for Apache Flink que usam o VVR 2.1.5 ou posterior.

Criar uma tabela sink

DDL mínimo

O exemplo a seguir mostra 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

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 é fixa.

Posição	Nome da coluna	Tipo	Obrigatório	Mapeia para
0	`metric`	VARCHAR	Sim	Nome do 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	Campo do InfluxDB (valor de dado)

Para gravar em várias colunas de campo, defina-as usando 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 causa 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 VPC da instância TSDB for InfluxDB. HTTP e 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 Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`password`	Sim	—	String	Senha do usuário especificado. Consulte Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`batchSize`	Não	`300`	Integer	Número de registros a gravar 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á usada. Consulte Manage user accounts and databasesManage user accounts and databasesManage user accounts and databasesManage user accounts and databases.
`ignoreErrorData`	Não	`false`	Boolean	Como tratar erros de gravação. `true`: ignora erros de gravação e continua. `false` (padrão): não ignora dados anormais.

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 a seguir 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)