Buat tabel sink InfluxDB dan contoh kode - Realtime Compute for Apache Flink

Penting

Konektor TSDB for InfluxDB akan ditinggalkan mulai versi 11.7. Setelah penghentian, konektor tersebut akan dihapus dari Konsol dan tidak lagi menerima pembaruan fitur atau pemeliharaan. Untuk informasi jadwal penghentian dukungan, lihat Akhir dukungan untuk konektor TSDB for InfluxDB. Segera migrasikan beban kerja Anda untuk menghindari gangguan pada pekerjaan produksi.

Konektor TSDB for InfluxDB menulis data streaming dari tabel sink Flink SQL ke instans TSDB for InfluxDB dalam Ververica Runtime (VVR). TSDB for InfluxDB adalah database deret waktu yang dioptimalkan untuk throughput tulis dan kueri tinggi, umumnya digunakan untuk pemantauan DevOps, metrik aplikasi, dan data sensor IoT.

Kemampuan konektor

Item	Nilai
Jenis tabel	Sink
Mode eksekusi	Streaming
Format data	Point
Jenis API	SQL
Pembaruan atau penghapusan data pada tabel sink	Tidak didukung
Metrik	`numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`

Untuk detail mengenai metrik ini, lihat Metrik pemantauan.

Prasyarat

Sebelum memulai, pastikan Anda telah:

Database yang dibuat di TSDB for InfluxDB. Lihat Mengelola akun pengguna dan databaseMengelola akun pengguna dan databaseMengelola akun pengguna dan databaseMengelola akun pengguna dan database.

Batasan

Konektor TSDB for InfluxDB hanya didukung oleh penerapan Realtime Compute for Apache Flink yang menggunakan VVR 2.1.5 atau lebih baru.

Buat tabel sink

DDL minimal

Contoh berikut menunjukkan kolom minimum yang diperlukan untuk mendefinisikan tabel 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>'
);

Konvensi kolom skema

Kolom tabel sink harus mengikuti konvensi penamaan tetap yang dipetakan ke model data InfluxDB. Urutan kolom bersifat tetap.

Posisi	Nama kolom	Tipe	Wajib	Dipetakan ke
0	`metric`	VARCHAR	Ya	Nama measurement InfluxDB
1	`timestamp`	BIGINT	Ya	Timestamp InfluxDB; satuan harus dalam milidetik
2+	`tag_<name>`	VARCHAR	Minimal satu	Tag InfluxDB (metadata terindeks)
3+	`field_<name>`	Tipe apa pun yang didukung	Minimal satu	Field InfluxDB (nilai data)

Untuk menulis ke beberapa kolom field, definisikan dengan pola berikut:

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

Catatan

Hanya nama kolom metric, timestamp, tag_*, dan field_* yang didukung. Nama kolom lainnya menyebabkan error.

Opsi konektor

Parameter	Wajib	Bawaan	Tipe	Deskripsi
`connector`	Ya	—	String	Harus `influxdb`.
`url`	Ya	—	String	Titik akhir VPC instans TSDB for InfluxDB. HTTP dan HTTPS didukung. Contoh: `https://localhost:8086` atau `http://localhost:3242`.
`database`	Ya	—	String	Nama database. Contoh: `db-flink`.
`username`	Ya	—	String	Username untuk database. Pengguna harus memiliki izin tulis pada database target. Lihat Kelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan database.
`password`	Ya	—	String	Password untuk pengguna yang ditentukan. Lihat Kelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan database.
`batchSize`	Tidak	`300`	Integer	Jumlah catatan yang ditulis dalam satu batch.
`retentionPolicy`	Tidak	`autogen`	String	Kebijakan retensi database target. Jika tidak ditentukan, kebijakan retensi bawaan database (`autogen`) digunakan. Lihat Kelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan databaseKelola akun pengguna dan database.
`ignoreErrorData`	Tidak	`false`	Boolean	Cara menangani error tulis. `true`: abaikan error tulis dan lanjutkan. `false`: hentikan pekerjaan jika terjadi error tulis.

Pemetaan tipe data

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

Contoh

Contoh berikut menghasilkan data acak menggunakan konektor datagen dan menuliskannya ke 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® (sedang ditinggalkan)