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}',            -- キーの漏洩を防ぐため、AccessKey ペアには変数を使用します。
  'password'='${secret_values.ak_secret}',        
  'endpoint'='<yourEndpoint>',
  'source.binlog.change-log-mode'='ALL',  -- INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER を含むすべてのチェンジログタイプを消費します。
  'retry-count'='10',                     -- バイナリログの読み取りエラーに対するリトライ回数。
  'retry-sleep-step-ms'='5000',           -- リトライ間のインクリメンタルバックオフ時間。最初のリトライは 5 秒、2 回目は 10 秒待機し、以降同様です。
  'source.binlog.batch-size'='512'        -- バイナリログデータを消費するためのバッチサイズ。
);

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}',       -- キーの漏洩を防ぐため、AccessKey ペアには変数を使用します。   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'cdcMode' = 'true',
  'sdkMode'='jdbc',
  'binlogMaxRetryTimes' = '10',     -- バイナリログの読み取りエラーに対するリトライ回数。
  'binlogRetryIntervalMs' = '500',  -- バイナリログの読み取りエラー後の再試行間隔 (ミリ秒)。
  'binlogBatchReadSize' = '100'     -- バイナリログデータを消費するためのバッチサイズ。
);

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}',       -- キーの漏洩を防ぐため、AccessKey ペアには変数を使用します。   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'source.binlog.change-log-mode'='ALL_AS_APPEND_ONLY',  -- すべてのチェンジログタイプを INSERT 操作として扱います。
  'retry-count'='10',                     -- バイナリログの読み取りエラーに対するリトライ回数。
  'retry-sleep-step-ms'='5000',           -- リトライ間のインクリメンタルバックオフ時間。最初のリトライは 5 秒、2 回目は 10 秒待機し、以降同様です。
  'source.binlog.batch-size'='512'        -- バイナリログデータを消費するためのバッチサイズ。
);

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}',       -- キーの漏洩を防ぐため、AccessKey ペアには変数を使用します。   
  'password' = '${secret_values.ak_secret}',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',     -- バイナリログの読み取りエラーに対するリトライ回数。
  'binlogRetryIntervalMs' = '500',  -- バイナリログの読み取りエラー後の再試行間隔 (ミリ秒)。
  'binlogBatchReadSize' = '100'     -- バイナリログデータを消費するためのバッチサイズ。
);

VVR 11+

次の DDL で Hologres パーティションテーブルを作成し、バイナリログと動的パーティショニングが有効になっていると仮定します。

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',   -- 動的パーティショニングを有効にします。
    auto_partitioning_time_unit = 'DAY',  -- パーティションは毎日作成されます。パーティション名の例: test_message_src1_20250512, test_message_src1_20250513。
    auto_partitioning_num_precreate = '2' -- 2 つのパーティションを事前作成します。
);
-- 既存のパーティションテーブルの場合、ALTER TABLE を使用して動的パーティショニングを有効にすることもできます。

Flink では、次の SQL ステートメントを使用して、パーティションテーブル test_message_src1 を DYNAMIC モードで消費します。

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  dt VARCHAR  -- Hologres パーティションテーブルのパーティション列。
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src1',  -- 動的パーティショニングが有効になっている親テーブル。
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'source.binlog.partition-binlog-mode' = 'DYNAMIC', -- 最新のパーティションを動的に監視します。
  'source.binlog.startup-mode' = 'initial'           -- すべての既存データを消費してから、バイナリログから増分データを消費します。
);

VVR 8.0.11

次の DDL で Hologres パーティションテーブルを作成し、バイナリログと動的パーティショニングが有効になっていると仮定します。

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',   -- 動的パーティショニングを有効にします。
    auto_partitioning_time_unit = 'DAY',  -- パーティションは毎日作成されます。パーティション名の例: test_message_src1_20241027, test_message_src1_20241028。
    auto_partitioning_num_precreate = '2' -- 2 つのパーティションを事前作成します。
);

-- 既存のパーティションテーブルの場合、ALTER TABLE を使用して動的パーティショニングを有効にすることもできます。

Flink では、次の SQL ステートメントを使用して、パーティションテーブル test_message_src1 のデータを DYNAMIC モードで消費します。

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  dt VARCHAR  -- Hologres パーティションテーブルのパーティション列。
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src1',  -- 動的パーティショニングが有効になっている親テーブル。
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'binlog' = 'true',
  'partition-binlog.mode' = 'DYNAMIC',  -- 最新のパーティションを動的に監視します。
  'binlogstartUpMode' = 'initial',      -- すべての既存データを消費してから、バイナリログから増分データを消費します。
  'sdkMode' = 'jdbc_fixed'              -- 接続制限の問題を回避するため、このモードを使用します。
);

VVR 11+

次の DDL で Hologres パーティションテーブルを作成し、バイナリログが有効になっていると仮定します。

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');

Flink では、次の SQL ステートメントを使用して、パーティションテーブル test_message_src2 を STATIC モードで消費します。

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  color VARCHAR  -- Hologres パーティションテーブルのパーティション列。
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src2',  -- パーティションテーブル。
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'source.binlog.partition-binlog-mode' = 'STATIC', -- 固定されたパーティションセットを消費します。
  'source.binlog.partition-values-to-read' = 'red,blue,green',  -- 指定された 3 つのパーティションのみを消費します。'black' パーティションは消費されません。新しいパーティションも消費されません。このオプションを指定しない場合、ジョブは親テーブルのすべてのパーティションを消費します。
  'source.binlog.startup-mode' = 'initial'  -- すべての既存データを消費してから、バイナリログから増分データを消費します。
);

VVR 8.0.11

次の DDL で Hologres パーティションテーブルを作成し、バイナリログが有効になっていると仮定します。

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');

Flink では、次の SQL ステートメントを使用して、パーティションテーブル test_message_src2 のデータを STATIC モードで消費します。

CREATE TEMPORARY TABLE hologres_source
(
  id INTEGER,
  title VARCHAR,
  body VARCHAR,
  color VARCHAR  -- Hologres パーティションテーブルのパーティション列。
)
with (
  'connector' = 'hologres',
  'dbname' = '<yourDatabase>',
  'tablename' = 'test_message_src2',  -- パーティションテーブル。
  'username' = '<yourUserName>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>',
  'binlog' = 'true',
  'partition-binlog.mode' = 'STATIC', -- 固定されたパーティションセットを消費します。
  'partition-values-to-read' = 'red,blue,green',  -- 指定された 3 つのパーティションのみを消費します。'black' パーティションは消費されません。新しいパーティションも消費されません。このオプションを指定しない場合、ジョブは親テーブルのすべてのパーティションを消費します。
  'binlogstartUpMode' = 'initial',  -- すべての既存データを消費してから、バイナリログから増分データを消費します。
  'sdkMode' = 'jdbc_fixed' -- 接続制限の問題を回避するため、このモードを使用します。
);

VVR 11.3+

public class Sample {                                                                                                                                                                          
        public static void main(String[] args) throws Exception {
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 読み取るテーブルのスキーマを初期化します。スキーマは Hologres テーブルスキーマのフィールドと一致する必要があります。フィールドのサブセットを定義できます。
            TableSchema schema = TableSchema.builder()
                    .field("a", DataTypes.INT())
                    .field("b", DataTypes.STRING())
                    .field("c", DataTypes.TIMESTAMP())
                    .build();

            // 読み取るテーブルの名前。
            String sourceTableName = "sourceTableName";

            // 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, sourceTableName);
            config.set(HologresConfigs.BINLOG, true);
            config.set(HologresConfigs.BINLOG_CHANGE_LOG_MODE, BinlogChangeLogMode.ALL);
            // Hologres Binlog ソースをビルドします。
            HologresBinlogSource source = new HologresBinlogSource(
                    new HologresConnectionParam(config),
                    schema,
                    config,
                    StartupMode.INITIAL,
                    sourceTableName,
                    "",
                    Collections.emptyList(),
                    -1,
                    Collections.emptySet(),
                    Collections.emptyList()
            );
            env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
            env.execute();
        }
  }

VVR 8.0.11+

public class Sample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 読み取るテーブルのスキーマを初期化します。スキーマは Hologres テーブルスキーマのフィールドと一致する必要があります。フィールドのサブセットを定義できます。
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .field("b", DataTypes.STRING())
                .field("c", DataTypes.TIMESTAMP())
                .build();
        // 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);
        // JDBCOptions をビルドします。
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Hologres Binlog ソースをビルドします。
        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();
        // 読み取るテーブルのスキーマを初期化します。スキーマは Hologres テーブルスキーマのフィールドと一致する必要があります。フィールドのサブセットを定義できます。
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .field("b", DataTypes.STRING())
                .field("c", DataTypes.TIMESTAMP())
                .build();
        // 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);
        // JDBCOptions をビルドします。
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Hologres Binlog ソースをビルドします。
        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();
        // 読み取るテーブルのスキーマを初期化します。スキーマは Hologres テーブルスキーマのフィールドと一致する必要があります。フィールドのサブセットを定義できます。
        TableSchema schema = TableSchema.builder()
                .field("a", DataTypes.INT())
                .build();
         // 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);
        // JDBCOptions をビルドします。
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // デフォルトのスロット名を設定または作成します。
        config.setString(HologresBinlogConfigs.JDBC_BINLOG_SLOT_NAME, HoloBinlogUtil.getOrCreateDefaultSlotForJDBCBinlog(jdbcOptions));

        boolean cdcMode = config.get(HologresBinlogConfigs.BINLOG_CDC_MODE) && config.get(HologresBinlogConfigs.OPTIONAL_BINLOG);
        // JDBCBinlogRecordConverter をビルドします。
        JDBCBinlogRecordConverter recordConverter = new JDBCBinlogRecordConverter(
                jdbcOptions.getTable(),
                schema,
                new HologresConnectionParam(config),
                cdcMode,
                Collections.emptySet());
        
        // Hologres Binlog ソースをビルドします。
        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();
    }
}