Buat saluran data - Tablestore SDK for Java - Tablestore

Gunakan Tablestore SDK for Java untuk membuat saluran data bertipe BaseData, Stream, atau BaseAndStream pada tabel data.

Prasyarat

Deskripsi fitur

Panggil createTunnel untuk membuat saluran data pada tabel data. Anda dapat membuat beberapa saluran data pada tabel yang sama. Tipe saluran data menentukan jenis data yang dikonsumsi: BaseData hanya mengonsumsi data lengkap, Stream hanya mengonsumsi data inkremental, dan BaseAndStream mengonsumsi data lengkap terlebih dahulu, diikuti oleh data inkremental.

Catatan

Jika Stream dinonaktifkan pada tabel data, pembuatan saluran data bertipe Stream atau BaseAndStream akan secara otomatis mengaktifkan Stream dan mengatur periode retensi log inkremental menjadi tujuh hari.

CreateTunnelResponse createTunnel(CreateTunnelRequest request)
        throws TableStoreException, ClientException

Kode contoh berikut membuat saluran data bertipe BaseData bernama example_tunnel pada tabel data example_table.

String tableName = "example_table";
String tunnelName = "example_tunnel";

CreateTunnelRequest request =
        new CreateTunnelRequest(
                tableName, tunnelName, TunnelType.BaseData);
CreateTunnelResponse response = tunnelClient.createTunnel(request);

System.out.println("TunnelId: " + response.getTunnelId());

Parameter

CreateTunnelRequest mencakup parameter berikut:

Nama	Tipe	Deskripsi
tableName (wajib)	`String`	Nama tabel data.
tunnelName (wajib)	`String`	Nama saluran data.
tunnelType (wajib)	`TunnelType`	Tipe saluran data. `BaseData` hanya mengonsumsi data lengkap. `Stream` hanya mengonsumsi data inkremental. `BaseAndStream` mengonsumsi data lengkap terlebih dahulu lalu dilanjutkan dengan data inkremental.
streamTunnelConfig (opsional)	`StreamTunnelConfig`	Konfigurasi rentang data inkremental untuk saluran data tipe `Stream` dan `BaseAndStream`. Jika Anda mengonfigurasi parameter ini, tentukan rentang waktu yang valid.
streamRecordOptions (opsional)	`StreamRecordOptions`	Konfigurasi catatan inkremental untuk saluran data tipe `Stream` dan `BaseAndStream`. Parameter ini memerlukan Tablestore SDK for Java versi 5.17.11 atau lebih baru.

Rentang data inkremental

streamTunnelConfig bertipe StreamTunnelConfig dan mencakup parameter berikut:

Nama	Tipe	Deskripsi
flag (opsional)	`StartOffsetFlag`	Posisi awal yang digunakan ketika `startOffset` tidak ditentukan. `LATEST` dimulai dari waktu pembuatan saluran data. `EARLIEST` dimulai dari log inkremental paling awal yang tersedia. Nilai default: `LATEST`.
startOffset (opsional)	`long`	Timestamp awal data inkremental. Satuan: milidetik. Rentang valid: [Waktu sistem saat ini – periode retensi Stream + 5 menit, waktu sistem saat ini). Jika parameter ini ditentukan, `flag` tidak berlaku.
endOffset (opsional)	`long`	Timestamp akhir data inkremental. Satuan: milidetik. Jika kedua timestamp ditentukan, parameter ini harus lebih besar dari `startOffset`. Jika parameter ini tidak ditentukan, konsumsi data inkremental akan berlangsung terus-menerus.

Catatan

Periode retensi Stream adalah periode retensi log inkremental. Nilai maksimumnya adalah tujuh hari. Anda dapat mengatur periode ini saat mengaktifkan Stream pada tabel data, namun tidak dapat diubah setelah ditetapkan.

Konten catatan inkremental

streamRecordOptions bertipe StreamRecordOptions dan mencakup parameter berikut:

Nama	Tipe	Deskripsi
getVersionGeneratorValue (opsional)	`boolean`	Menentukan apakah catatan inkremental berisi nilai yang dihasilkan oleh generator versi. Nilai default: `false`.
getSysColumns (opsional)	`boolean`	Menentukan apakah catatan inkremental berisi kolom sistem. Nilai default: `false`.
getNewRowInfo (opsional)	`boolean`	Menentukan apakah catatan inkremental berisi informasi baris terbaru. Nilai default: `false`.
oldColumnsToGet (opsional)	`StreamColumn`	Kolom atribut yang dikembalikan dari baris asli.
newColumnsToGet (opsional)	`StreamColumn`	Kolom atribut yang dikembalikan dari baris terbaru.

Kolom catatan inkremental

streamRecordOptions.oldColumnsToGet dan streamRecordOptions.newColumnsToGet bertipe StreamColumn dan mencakup parameter berikut:

Nama	Tipe	Deskripsi
columnType (wajib)	`StreamColumnType`	Mode pemilihan kolom atribut. `SPECIFIED_COLUMN` menentukan kolom atribut berdasarkan nama. `INPUT_COLUMNS` menentukan kolom atribut yang ditulis atau diperbarui oleh operasi. `ALL_COLUMNS` menentukan semua kolom atribut.
columnNames (opsional)	`List<String>`	Nama kolom atribut yang dikembalikan ketika `columnType` diatur ke `SPECIFIED_COLUMN`.

Tanggapan

CreateTunnelResponse mencakup bidang tanggapan berikut:

Bidang	Type	Deskripsi
tunnelId	`String`	ID saluran data yang dibuat. Panggil `getTunnelId()` untuk mendapatkan ID tersebut. ID ini diperlukan untuk mengonsumsi data saluran data.

Skenario

Tentukan rentang data inkremental

Kode contoh berikut membuat saluran data bertipe Stream dan menentukan rentang data inkremental selama satu jam terakhir.

long endTime = System.currentTimeMillis() - 1_000L;
long startTime = endTime - 60 * 60 * 1000L;
StreamTunnelConfig streamConfig =
        new StreamTunnelConfig(startTime, endTime);

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table",
                "example_stream_tunnel",
                TunnelType.Stream);
request.setStreamTunnelConfig(streamConfig);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());

Konfigurasikan konten catatan inkremental

Kode contoh berikut membuat saluran data bertipe Stream dan mengonfigurasi catatan inkremental agar mengembalikan nilai generator versi, kolom sistem, informasi baris terbaru, kolom atribut value dari baris asli, serta semua kolom atribut dari baris terbaru.

StreamColumn oldColumns =
        new StreamColumn(StreamColumnType.SPECIFIED_COLUMN);
oldColumns.addColumnName("value");

StreamRecordOptions recordOptions = new StreamRecordOptions();
recordOptions.setGetVersionGeneratorValue(true);
recordOptions.setGetSysColumns(true);
recordOptions.setGetNewRowInfo(true);
recordOptions.setOldColumnsToGet(oldColumns);
recordOptions.setNewColumnsToGet(
        new StreamColumn(StreamColumnType.ALL_COLUMNS));

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table", "example_stream_tunnel", TunnelType.Stream);
request.setStreamRecordOptions(recordOptions);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());