实时计算Flink版：实时数仓Hologres

背景信息

实时数仓Hologres是一站式实时数据仓库引擎，支持海量数据实时写入、实时更新、实时分析，支持标准SQL（兼容PostgreSQL协议），支持PB级数据多维分析（OLAP）与即席分析（Ad Hoc），支持高并发低延迟的在线数据服务（Serving），与MaxCompute、Flink、DataWorks深度融合，提供离在线一体化全栈数仓解决方案。

特色功能

前提条件

已创建完成Hologres表，详情请参见创建Hologres表。

使用限制

• 通用：

  • 仅Flink计算引擎VVR 2.0.0及以上版本支持Hologres连接器。

  • Hologres连接器不支持访问Hologres外部表。关于Hologres外部表详情请参见外部表。

  • Hologres 2.0版本起下线RPC（用于维表与结果表）、Holohub模式（用于Binlog源表），全面转为jdbc相关模式（目前包括jdbc、jdbc_fixed、jdbc_copy等）。推荐维表和结果表使用jdbc_fixed模式，该模式不占用Hologres实例连接数，详情请参见WITH参数。实时计算VVR 6.0.6以及之前的版本，如果读写Hologres 2.0版本并使用了上述已经下线的两种模式可能抛出异常。

• 源表独有：

  • Flink默认以批模式读取Hologres源表数据，即只扫描一次Hologres全表，扫描结束，消费结束，新到Hologres源表的数据不会被读取。从VVR 3.0.0版本开始，支持实时消费Hologres数据。关于实时消费Hologres，详情请参见Flink实时消费Hologres。从VVR 6.0.3版本开始，Hologres源表在批模式读取时支持filter下推，详见源表独有参数enable_filter_push_down。

  • Hologres CDC模式暂不支持定义Watermark。如果您需要进行窗口聚合，您可以采用非窗口聚合的方式，详情请参见上下游存储。

• 结果表独有：

  无。

• 维表独有：

  • 创建Hologres维表时建议选择行存模式，列存模式对于点查场景性能开销较大。选择行存模式创建维表时必须设置主键，并且将主键设置为Clustering Key才可以工作。详情请参见建表概述。

  • Hologres维表的主键必须是Flink Join On的字段，Flink Join On的字段也必须是维表完整的主键字段，两者必须完全匹配。

  • Flink实时计算引擎为VVR 4.0及以上版本的Hologres Flink连接器支持非主键的维表JOIN，而VVR 4.0以下版本的Hologres Flink连接器仅支持主键的维表JOIN。

语法结构

CREATE TABLE hologres_table (
  name VARCHAR,
  age BIGINT,
  birthday BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'hologres',
  'dbname' = '<yourDBName>',
  'tablename' = '<yourTableName>',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'endpoint' = '<yourEndpoint>'
);

WITH参数

类型映射

Flink与Hologres的数据类型映射请参见Blink/Flink与Hologres的数据类型映射。

使用示例

源表示例

• 非Binlog Source表

  CREATE TEMPORARY TABLE hologres_source (
    name varchar, 
    age BIGINT,
    birthday BIGINT
  ) WITH (
    'connector'='hologres',
    'dbname'='<yourDbname>',
    'tablename'='<yourTablename>',
    'username'='<yourAccessID>',
    'password'='<yourAccessSecret>',
    'endpoint'='<yourEndpoint>',
    'field_delimiter'='|' --该参数可选。
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    name varchar,
    age BIGINT,
    birthday BIGINT 
  ) WITH (
    'connector'='blackhole'
  );
  
  INSERT INTO blackhole_sink SELECT 
     name, age, birthday
  from hologres_source;

• Binlog Source表

  Hologres连接器支持实时消费Hologres。实时消费Hologres，即实时消费Hologres的Binlog数据。以下为您介绍Flink实时消费Hologres的详情：

  • 使用限制

    • Hologres 0.10及以下版本，已存在的表无法修改表属性开启Binlog，需要重新建表。Hologres V1.1及以上版本，可以根据业务需要选择开启或关闭Binlog能力，同时支持配置TTL满足不同业务场景对Binlog保留时间的诉求，详情请参见订阅Hologres Binlog。

    • 不支持开启分区表父表的Binlog，请使用非分区表。

    • 暂不支持实时消费TIMESTAMP类型的数据，因此创建Hologres表时，请使用TIMESTAMPTZ类型。

    • 开启Binlog后，理论上列存表的开销要大于行存表的开销。因此，对于数据更新频繁的场景，建议使用存储格式为行存的表开启Binlog。

    • 默认的Binlog源表不支持数组类型，仅支持以下数据类型：

      • INTEGER

      • BIGINT

      • TEXT

      • REAL

      • DOUBLE PRECISION

      • BOOLEAN

      • NUMERIC(38,8)

      • TIMESTAMPTZ

      说明

      对不支持的数据类型（例如SMALLINT），即使不消费此字段，仍然可能导致作业无法上线。

    • 实时计算引擎VVR 6.0.3及以上版本新增JDBC模式Binlog源表。相比原有Holohub模式，支持更完善的数据类型，如SMALLINT，数组类型等，同时也支持了自定义用户（非RAM用户）。详见下方JDBC模式Binlog源表。

    • Hologres 1.3.41版本开始，JDBC模式Binlog源表新支持读取JSONB类型，但需要数据库级别开启GUC，开启GUC的命令如下。

      --db级别开启GUC，仅superuser可以执行，每个db只需要设置一次。
      alter database <db_name> set hg_experimental_enable_binlog_jsonb = on;

    • Hologres 2.0版本起下线Holohub模式，全面转为jdbc模式。

  • 开启Binlog

    实时消费功能默认关闭，因此在Hologres控制台上创建表的DDL时，需要设置binlog.level和binlog.ttl参数，示例如下。

    begin;
    CREATE TABLE test_message_src(
     id int primary key, 
     title text not null, 
     body text
    );
    call set_table_property('test_message_src', 'orientation', 'row');
    call set_table_property('test_message_src', 'clustering_key', 'id');
    call set_table_property('test_message_src', 'binlog.level', 'replica'); --自Hologres 1.1版本起，可以在建表后开启Binlog。
    call set_table_property('test_message_src', 'binlog.ttl', '86400'); 
    commit;

    其中，binlog.level设置为replica即代表开启Binlog，binlog.ttl为Binlog的TTL，单位为秒。

  • 注意事项

    • UPDATE操作会产生两条Binlog记录，分别为更新操作前和操作后的数据记录，因此您会消费到两条数据。但是，Hologres Binlog功能会保证这两条记录是连续的且更新前的Binlog记录在前，更新后的Binlog记录在后。

    • 建议Flink作业并发数和Hologres Table的Shard个数保持一致。

      您可以在Hologres控制台上，使用以下语句查看Table的Shard数，其中tablename为您的业务表名称。

      select tg.property_value from hologres.hg_table_properties tb join hologres.hg_table_group_properties tg on tb.property_value = tg.tablegroup_name where tb.property_key = 'table_group' and tg.property_key = 'shard_count' and table_name = '<tablename>';

  • 消费模式

    • 非CDC模式

      该模式下Source消费的Binlog数据是作为普通的Flink数据传递给下游节点的，即所有数据都是INSERT类型的数据，您可以根据业务情况选择如何处理特定hg_binlog_event_type类型的数据。源表DDL代码示例如下。

      CREATE 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'='<yourAccessID>',
        'password'='<yourAccessSecret>',
        'endpoint'='<yourEndpoint>',
        'binlog' = 'true',
        'binlogMaxRetryTimes' = '10',
        'binlogRetryIntervalMs' = '500',
        'binlogBatchReadSize' = '100'
      );

    • CDC模式

      该模式下Source消费的Binlog数据，将根据hg_binlog_event_type自动为每行数据设置准确的Flink RowKind类型，例如，INSERT、DELETE、UPDATE_BEFORE和UPDATE_AFTER类型，这样就能完成表的数据的镜像同步，类似MySQL或Postgres的CDC功能。源表DDL代码示例如下。

      CREATE TABLE test_message_src_binlog_table(
        id INTEGER,
        title VARCHAR,
        body VARCHAR
      ) WITH (
        'connector'='hologres',
        'dbname'='<yourDbname>',
        'tablename'='<yourTablename>',
        'username'='<yourAccessID>',
        'password'='<yourAccessSecret>',
        'endpoint'='<yourEndpoint>',
        'binlog' = 'true',
        'cdcMode' = 'true',
        'binlogMaxRetryTimes' = '10',
        'binlogRetryIntervalMs' = '500',
        'binlogBatchReadSize' = '100'
      );

  • 全增量一体源表消费：

    • 背景信息

      在源表Join维表时，由于Binlog的TTL等原因，会导致无法使用源表的所有数据。原解决方案是为Binlog表设置一个很大的TTL，但这样会有以下问题：

      • 历史Binlog数据会被长时间保存，导致占用较多的存储资源。

      • 因为Binlog包含数据更新记录，使用Binlog进行全量消费会消费一些不必要的数据，导致占用较多的计算资源，且无法让用户只关注最新的数据。

      从VVR 4.0.13及以上版本，Hologres 0.10及以上版本，Hologres Binlog CDC源表支持全增量一体的消费，这种方式会先读取数据库的历史全量数据，并平滑切换到Binlog读取增量数据。采用这种方式，可以解决上述问题。

    • 适用场景

      • 适用于历史数据不包含Binlog，但又希望消费所有数据的场景。

      • 仅适用于目标表有主键的场景，推荐在CDC模式下使用的全增量Hologres源表。

      • Hologres1.1版本之后，支持按需开启Binlog，可以将已有历史数据的表打开Binlog。

    • 代码示例

      CREATE 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'='<yourAccessID>',
        'password'='<yourAccessSecret>',
        'endpoint'='<yourEndpoint>',
        'binlog' = 'true',
        'cdcMode' = 'true',
        'binlogStartUpMode' = 'initial', --先读取历史全量数据，再增量消费Binlog。
        'binlogMaxRetryTimes' = '10',
        'binlogRetryIntervalMs' = '500',
        'binlogBatchReadSize' = '100'
        );

  • JDBC模式Binlog源表

    • 实时计算引擎VVR 6.0.3版本开始，binlog源表新增JDBC模式（不同于CDC等消费模式，此处的JDBC模式是指底层获取binlog的SDK基于JDBC）。相比原有Holohub模式，支持更多的数据类型，包括：SMALLINT、INTEGER、BIGINT、TEXT、REAL、DOUBLE PRECISION、BOOLEAN、NUMERIC、DATE、TIME、TIMETZ、TIMESTAMP、TIMESTAMPTZ、BYTEA、JSON、int4[]、int8[]、float4[]、float8[]、boolean[]、text[]、JSONB（需要Hologres版本大于1.3.41且开启相应GUC，详见上方Binlog Source源表使用限制）。同时，JDBC模式的Binlog源表支持Hologres的自定义用户（非RAM用户）。

    • 使用方式与普通的binlog源表类似，但需要设置sdkMode为jdbc，示例如下：

      create TEMPORARY table test_message_src_binlog_table(
        id INTEGER,
        title VARCHAR,
        body VARCHAR
      ) with (
        'connector'='hologres',
        'dbname'='<yourDbname>',
        'tablename'='<yourTablename>',
        'username'='<yourAccessID>',
        'password'='<yourAccessSecret>',
        'endpoint'='<yourEndpoint>',
        'binlog' = 'true',
        'cdcMode' = 'true',
        'sdkMode'='jdbc', --使用jdbc模式的binlog源表
        'jdbcBinlogSlotName'='replication_slot_name' --jdbc模式的binlog源表需要设置slot name
      );
      

    • 实时计算引擎VVR 6.0.3到VVR 6.0.5版本使用JDBC模式消费binlog需要一些准备工作，包括创建publication、replication_slot、源表的DDL中必须设置jdbcBinlogSlotName参数。相关概念以及执行所需权限的详情可以参见通过JDBC消费Hologres Binlog。具体的SQL如下，需要在hologres中执行。

      --创建extension, 为db级别开启，仅superuser可以执行，每个db只需要设置一次，Hologres 2.0版本开始默认开启。
      create extension if not exists hg_binlog;
      
      --创建publication，对应具体的表。进行此操作需要目标表的Owner权限。
      create publication <publication_name> for table <table_name>;
      
      --创建logical_replication_slot，对应具体的publication。进行此操作需要实例的Superuser或Replication Role权限。
      call hg_create_logical_replication_slot('<replication_slot_name>', 'hgoutput', '<publication_name>');

    • 实时计算引擎VVR 6.0.6版本开始，可以选择不设置jdbcBinlogSlotName，Hologres连接器会创建默认的slot并使用。但默认创建slot需要一定的前提条件，要求用户为实例的Superuser，或用户同时拥有目标表的Owner权限和实例的Replication Role权限。相关概念以及授权操作的详情可以参见通过JDBC消费Hologres Binlog。

  • Hologres Binlog实现原理

    一条Binlog的字段由Binlog系统字段和用户Table字段组成，字段定义如下：

    字段名	字段类型	说明
    hg_binlog_lsn	BIGINT	Binlog系统字段，表示Binlog序号，Shard内部单调递增不保证连续，不同Shard之间不保证唯一和有序。
    hg_binlog_event_type	BIGINT	Binlog系统字段，表示当前记录所表示的修改类型，参数取值如下： INSERT=5：表示当前Binlog为插入一条新的记录。 DELETE=2：表示当前Binlog为删除一条已有的记录。 BEFORE_UPDATE=3：表示当前Binlog为更新操作前的记录。 AFTER_UPDATE=7：表示当前Binlog为更新操作后的记录。
    hg_binlog_timestamp_us	BIGINT	Binlog系统字段，系统时间戳，单位为微秒。
    user_table_column_1	用户定义	用户的表字段。
    ...	...	用户的表字段。
    user_table_column_n	用户定义	用户的表字段。

结果表示例

CREATE TEMPORARY TABLE datagen_source(
  name varchar, 
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE hologres_sink (
  name varchar, 
  age BIGINT,
  birthday BIGINT
) WITH (
  'connector'='hologres',           
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>'
);

INSERT INTO hologres_sink SELECT * from datagen_source;

维表示例

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  proctime AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE hologres_dim (
  a INT, 
  b VARCHAR, 
  c VARCHAR
) WITH (
  'connector' = 'hologres',
   ...
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink SELECT T.a,H.b
FROM datagen_source AS T JOIN hologres_dim FOR SYSTEM_TIME AS OF T.proctime AS H ON T.a = H.a;

特色功能详解

流式语义

流处理，也称为流数据或流事件处理，即对一系列无界数据或事件连续处理。执行流数据或流事件处理的系统通常允许您指定一种可靠性模式或处理语义，保证整个系统处理数据的准确性，因为网络或设备故障等可能会导致数据丢失。

宽表Merge和局部更新功能

在把多个流的数据写到一张Hologres宽表的场景中，会涉及到宽表Merge和数据的局部更新。下面通过示例来介绍如何设置。

假设有两个Flink数据流，一个数据流中包含A、B和C字段，另一个数据流中包含A、D和E字段，Hologres宽表WIDE_TABLE包含A、B、C、D和E字段，其中A字段为主键。具体操作如下：

1. 使用Flink SQL创建两张Hologres结果表，其中一张表只声明A、B和C字段，另一张表只声明A、D和E字段。这两张表都映射至宽表WIDE_TABLE。

2. 两张结果表的属性设置：

   • mutatetype设置为insertorupdate，可以根据主键更新数据。

   • ignoredelete设置为true，防止回撤消息产生Delete请求。

3. 将两个Flink数据流的数据分别INSERT至对应的结果表中。

作为CTAS和CDAS的目标端

Hologres支持实时同步单表或整库级别的数据，在同步过程之中如果上游的表结构发生了变更也会实时同步到Hologres表中。

当上游需要同步的表发生了表结构变更，随着新数据流到Hologres表时，Flink会先触发Hologres修改相应的表结构，然后再将数据写入到相应的表中。以上过程，全部由Flink自动完成，您不需要关心实现细节。详情请参见CREATE TABLE AS（CTAS）语句、CREATE DATABASE AS（CDAS）语句和数据库实时入仓快速入门。

DataStream API

• Hologres源表

  VVR提供了RichInputFormat的实现类HologresBulkreadInputFormat来读取Hologres表数据。以下为构建Hologres Source读取表数据的示例。

  // 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
  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");
  // 构建JDBC Options。
  JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
  String query = JDBCUtils.getSimpleSelectFromStatement(
      jdbcOptions.getTable(), schema.getFieldNames());
  
  // 构建HologresBulkreadInputFormat读取表数据。
  HologresBulkreadInputFormat inputFormat = new HologresBulkreadInputFormat(jdbcOptions, schema, query);
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
  env.addSource(new InputFormatSourceFunction<>(inputFormat, typeInfo)).returns(typeInfo)
      .print();
  env.execute();

• Hologres Binlog源表

  VVR提供了Source的实现类HologresBinlogSource来读取Hologres Binlog数据。以下为构建Hologres Binlog Source的示例。

  // 初始化读取的表的Schema，需要和Hologres表Schema的字段匹配，可以只定义部分字段。
  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.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
  // 构建JDBC Options。
  JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
  jdbcOptions.setHolohubEndpoint(JDBCUtils.getHolohubEndpoint(jdbcOptions));
  RowDataRecordConverter recordConverter = buildRecordConverter(schema, config, jdbcOptions);
  // 构建Hologres Binlog Source。
  long startTimeMs = 0;
  HologresBinlogSource<RowData> source = new HologresBinlogSource<>(
      schema,
      config,
      jdbcOptions,
      recordConverter,
      startTimeMs);
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
  env.execute();

  说明

  • 方法buildRecordConverter不在VVR Connector依赖中，是示例代码中提供的方法。

  • Hologres Binlog注意事项和实现原理等详情，请参见Binlog Source表。

• Hologres结果表

  VVR提供了OutputFormatSinkFunction的实现类HologresSinkFunction来写入数据。以下为构建Hologres Sink的示例。

  // 初始化读取的表的Schema。
  TableSchema schema = TableSchema.builder()
      .field("a", DataTypes.INT())
    .field("b", DataTypes.STRING())
      .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.setBoolean(HologresConfigs.USE_RPC_MODE, true);
  HologresConnectionParam hologresConnectionParam = new HologresConnectionParam(config);
  // 构建Hologres Writer，以RowData的方式写入数据。
  AbstractHologresWriter<RowData> hologresWriter =
      buildHologresWriter(schema, config, hologresConnectionParam);
  // 构建Hologres Sink。
  HologresSinkFunction sinkFunction = new HologresSinkFunction(hologresConnectionParam, hologresWriter);
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  TypeInformation<RowData> typeInfo = InternalTypeInfo.of(schema.toRowDataType().getLogicalType());
  int offset = (int) (System.currentTimeMillis() % Integer.MAX_VALUE);
  env.fromElements((RowData)GenericRowData.of(2 + offset, StringData.fromString("2")), GenericRowData.of(3 + offset, StringData.fromString("3"))).returns(typeInfo)
      .addSink(sinkFunction);
  env.execute();

  说明

  方法buildHologresWriter不在VVR Connector依赖中，是示例代码中提供的方法。