Realtime Compute for Apache Flink：Hologres

VVR 11+

CREATE TEMPORARY TABLE test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='${secret_values.ak_id}',            --Gunakan manajemen variabel untuk Pasangan Kunci Akses Anda guna mencegah kebocoran kunci. 
  'password'='${secret_values.ak_secret}',        
  'endpoint'='<yourEndpoint>',
  'source.binlog.change-log-mode'='ALL',  --Baca semua jenis ChangeLog, termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER.
  'retry-count'='10',                     --Jumlah percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'retry-sleep-step-ms'='5000',           --Waktu tunggu bertahap untuk percobaan ulang. Percobaan ulang pertama menunggu 5 detik, yang kedua 10 detik, dan seterusnya.
  'source.binlog.batch-size'='512'        --Jumlah baris data yang dibaca dari log biner dalam setiap batch.
);

VVR 8+

CREATE TEMPORARY TABLE test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username' = '${secret_values.ak_id}',       --Gunakan manajemen variabel untuk Pasangan Kunci Akses Anda guna mencegah kebocoran kunci.   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'cdcMode' = 'true',
  'sdkMode'='jdbc',
  'binlogMaxRetryTimes' = '10',     --Jumlah percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'binlogRetryIntervalMs' = '500',  --Interval percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'binlogBatchReadSize' = '100'     --Jumlah baris data yang dibaca dari log biner dalam setiap batch.
);

VVR 11+

CREATE TEMPORARY TABLE test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username' = '${secret_values.ak_id}',       --Gunakan manajemen variabel untuk Pasangan Kunci Akses Anda guna mencegah kebocoran kunci.   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'source.binlog.change-log-mode'='ALL_AS_APPEND_ONLY',  --Semua jenis ChangeLog diproses sebagai INSERT.
  'retry-count'='10',                     --Jumlah percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'retry-sleep-step-ms'='5000',           --Waktu tunggu bertahap untuk percobaan ulang. Percobaan ulang pertama menunggu 5 detik, yang kedua 10 detik, dan seterusnya.
  'source.binlog.batch-size'='512'        --Jumlah baris data yang dibaca dari log biner dalam setiap batch.
);

VVR 8+

CREATE TEMPORARY TABLE test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) WITH (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username' = '${secret_values.ak_id}',       --Gunakan manajemen variabel untuk Pasangan Kunci Akses Anda guna mencegah kebocoran kunci.   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',     --Jumlah percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'binlogRetryIntervalMs' = '500',  --Interval percobaan ulang setelah terjadi kesalahan pembacaan log biner.
  'binlogBatchReadSize' = '100'     --Jumlah baris data yang dibaca dari log biner dalam setiap batch.
);

VVR 11+

Asumsikan tabel partisi DDL berikut ada di Hologres, dan binary logging serta partisi dinamis diaktifkan.

CREATE TABLE "test_message_src1" (
    id int,
    title text,
    body text,
    dt text,
    PRIMARY KEY (id, dt)
)
PARTITION BY LIST (dt) WITH (
    binlog_level = 'replica', 
    auto_partitioning_enable =  'true',   -- Aktifkan partisi dinamis.
    auto_partitioning_time_unit = 'DAY',  -- Gunakan DAY sebagai satuan waktu. Contoh nama partisi: test_message_src1_20250512, test_message_src1_20250513.
    auto_partitioning_num_precreate = '2' -- Pra-buat dua partisi.
);
-- Untuk tabel partisi yang sudah ada, Anda juga dapat mengaktifkan partisi dinamis menggunakan ALTER TABLE.

Di Flink, Anda dapat menggunakan pernyataan SQL berikut untuk mendeklarasikan konsumsi mode DYNAMIC untuk tabel partisi test_message_src1.

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  dt VARCHAR  -- Field partisi dari tabel partisi Hologres.
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src1',  -- Tabel induk dengan partisi dinamis diaktifkan.
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'source.binlog.partition-binlog-mode' = 'DYNAMIC', -- Secara dinamis mendengarkan partisi terbaru.
  'source.binlog.startup-mode' = 'initial'           -- Pertama membaca semua data historis, lalu mengonsumsi log biner secara inkremental.
);

VVR 8.0.11

Asumsikan tabel partisi DDL berikut ada di Hologres, dan binary logging serta partisi dinamis diaktifkan.

CREATE TABLE "test_message_src1" (
    id int,
    title text,
    body text,
    dt text,
    PRIMARY KEY (id, dt)
)
PARTITION BY LIST (dt) WITH (
    binlog_level = 'replica', 
    auto_partitioning_enable =  'true',   -- Aktifkan partisi dinamis.
    auto_partitioning_time_unit = 'DAY',  -- Gunakan DAY sebagai satuan waktu. Contoh nama partisi: test_message_src1_20241027, test_message_src1_20241028.
    auto_partitioning_num_precreate = '2' -- Pra-buat dua partisi.
);

-- Untuk tabel partisi yang sudah ada, Anda juga dapat mengaktifkan partisi dinamis menggunakan ALTER TABLE.

Di Flink, Anda dapat menggunakan pernyataan SQL berikut untuk mendeklarasikan konsumsi mode DYNAMIC untuk tabel partisi test_message_src1.

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  dt VARCHAR  -- Field partisi dari tabel partisi Hologres.
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src1',  -- Tabel induk dengan partisi dinamis diaktifkan.
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'binlog' = 'true',
  'partition-binlog.mode' = 'DYNAMIC',  -- Secara dinamis mendengarkan partisi terbaru.
  'binlogstartUpMode' = 'initial',      -- Pertama membaca semua data historis, lalu mengonsumsi log biner secara inkremental.
  'sdkMode' = 'jdbc_fixed'              -- Gunakan mode ini untuk menghindari batas koneksi.
);

VVR 11+

Asumsikan tabel partisi DDL berikut ada di Hologres, dan binary logging diaktifkan.

CREATE TABLE test_message_src2 (
    id int,
    title text,
    body text,
    color text,
    PRIMARY KEY (id, color)
)
PARTITION BY LIST (color) WITH (
    binlog_level = 'replica'
);
create table test_message_src2_red partition of test_message_src2 for values in ('red');
create table test_message_src2_blue partition of test_message_src2 for values in ('blue');
create table test_message_src2_green partition of test_message_src2 for values in ('green');
create table test_message_src2_black partition of test_message_src2 for values in ('black');

Di Flink, Anda dapat menggunakan pernyataan SQL berikut untuk mendeklarasikan konsumsi mode STATIC untuk tabel partisi test_message_src2.

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  color VARCHAR  -- Field partisi dari tabel partisi Hologres.
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src2',  -- Tabel partisi.
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'source.binlog.partition-binlog-mode' = 'STATIC', -- Mengonsumsi partisi tetap.
  'source.binlog.partition-values-to-read' = 'red,blue,green',  -- Hanya mengonsumsi tiga partisi yang dikonfigurasi. Partisi 'black' tidak dikonsumsi. Partisi baru tidak akan dikonsumsi. Jika tidak diatur, semua partisi dari tabel induk dikonsumsi.
  'source.binlog.startup-mode' = 'initial'  -- Pertama membaca semua data historis, lalu mengonsumsi log biner secara inkremental.
);

VVR 8.0.11

Asumsikan tabel partisi DDL berikut ada di Hologres, dan binary logging diaktifkan.

CREATE TABLE test_message_src2 (
    id int,
    title text,
    body text,
    color text,
    PRIMARY KEY (id, color)
)
PARTITION BY LIST (color) WITH (
    binlog_level = 'replica'
);
create table test_message_src2_red partition of test_message_src2 for values in ('red');
create table test_message_src2_blue partition of test_message_src2 for values in ('blue');
create table test_message_src2_green partition of test_message_src2 for values in ('green');
create table test_message_src2_black partition of test_message_src2 for values in ('black');

Di Flink, Anda dapat menggunakan pernyataan SQL berikut untuk mendeklarasikan konsumsi mode STATIC untuk tabel partisi test_message_src2.

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  color VARCHAR  -- Field partisi dari tabel partisi Hologres.
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src2',  -- Tabel partisi.
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'binlog' = 'true',
  'partition-binlog.mode' = 'STATIC', -- Mengonsumsi partisi tetap.
  'partition-values-to-read' = 'red,blue,green',  -- Hanya mengonsumsi tiga partisi yang dikonfigurasi. Partisi 'black' tidak dikonsumsi. Partisi baru tidak akan dikonsumsi. Jika tidak diatur, semua partisi dari tabel induk dikonsumsi.
  'binlogstartUpMode' = 'initial',  -- Pertama membaca semua data historis, lalu mengonsumsi log biner secara inkremental.
  'sdkMode' = 'jdbc_fixed' -- Gunakan mode ini untuk menghindari batas koneksi.
);

VVR 8.0.11+

public class Sample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Inisialisasi skema tabel yang akan dibaca. Field harus sesuai dengan skema tabel Hologres. Anda hanya dapat mendefinisikan subset field.
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .field("b", DataTypes.STRING())
                .field("c", DataTypes.TIMESTAMP())
                .build();
        // Parameter terkait Hologres.
        Configuration config = new Configuration();
        config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
        config.setString(HologresConfigs.USERNAME, "yourUserName");
        config.setString(HologresConfigs.PASSWORD, "yourPassword");
        config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
        config.setString(HologresConfigs.TABLE, "yourTableName");
        config.setString(HologresConfigs.SDK_MODE, "jdbc");
        config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
        config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
        // Bangun JDBCOptions.
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Bangun HologresBinlogSource.
        long startTimeMs = 0;
        HologresBinlogSource source = new HologresBinlogSource(
                new HologresConnectionParam(config),
                schema,
                config,
                jdbcOptions,
                startTimeMs,
                StartupMode.INITIAL,
                "",
                "",
                -1,
                Collections.emptySet(),
                new ArrayList<>()
        );
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
        env.execute();
    }
}

VVR 8.0.7+

public class Sample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Inisialisasi skema tabel yang akan dibaca. Field harus sesuai dengan skema tabel Hologres. Anda hanya dapat mendefinisikan subset field.
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .field("b", DataTypes.STRING())
                .field("c", DataTypes.TIMESTAMP())
                .build();
        // Parameter terkait Hologres.
        Configuration config = new Configuration();
        config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
        config.setString(HologresConfigs.USERNAME, "yourUserName");
        config.setString(HologresConfigs.PASSWORD, "yourPassword");
        config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
        config.setString(HologresConfigs.TABLE, "yourTableName");
        config.setString(HologresConfigs.SDK_MODE, "jdbc");
        config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
        config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
        // Bangun JDBCOptions.
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Bangun HologresBinlogSource.
        long startTimeMs = 0;
        HologresBinlogSource source = new HologresBinlogSource(
                new HologresConnectionParam(config),
                schema,
                config,
                jdbcOptions,
                startTimeMs,
                StartupMode.INITIAL,
                "",
                "",
                -1,
                Collections.emptySet()
        );
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
        env.execute();
    }
}

VVR 6.0.7+

public class Sample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Inisialisasi skema tabel yang akan dibaca. Field harus sesuai dengan skema tabel Hologres. Anda hanya dapat mendefinisikan subset field.
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .build();
         // Parameter terkait Hologres.
        Configuration config = new Configuration();
        config.setString(HologresConfigs.ENDPOINT, "yourEndpoint");
        config.setString(HologresConfigs.USERNAME, "yourUserName");
        config.setString(HologresConfigs.PASSWORD, "yourPassword");
        config.setString(HologresConfigs.DATABASE, "yourDatabaseName");
        config.setString(HologresConfigs.TABLE, "yourTableName");
        config.setString(HologresConfigs.SDK_MODE, "jdbc");
        config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
        config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
        // Bangun JDBCOptions.
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Atur atau buat nama slot default.
        config.setString(HologresBinlogConfigs.JDBC_BINLOG_SLOT_NAME, HoloBinlogUtil.getOrCreateDefaultSlotForJDBCBinlog(jdbcOptions));

        boolean cdcMode = config.get(HologresBinlogConfigs.BINLOG_CDC_MODE) && config.get(HologresBinlogConfigs.OPTIONAL_BINLOG);
        // Bangun Konverter Catatan Binlog.
        JDBCBinlogRecordConverter recordConverter = new JDBCBinlogRecordConverter(
                jdbcOptions.getTable(),
                schema,
                new HologresConnectionParam(config),
                cdcMode,
                Collections.emptySet());
        
        // Bangun HologresBinlogSource.
        long startTimeMs = 0;
        HologresJDBCBinlogSource source = new HologresJDBCBinlogSource(
                new HologresConnectionParam(config),
                schema,
                config,
                jdbcOptions,
                startTimeMs,
                StartupMode.TIMESTAMP,
                recordConverter,
                "",
                -1);
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
        env.execute();
    }
}