实时数仓Hologres：通过JDBC消费Hologres Binlog

前提条件

• 需提前开启和配置Hologres Binlog，详情请参见订阅Hologres Binlog。

• 需创建hg_binlog extension。

  • Hologres V2.0版本前，需要实例的Superuser执行以下语句创建Extension才可以使用该功能，Extension针对整个DB生效，一个DB只需执行一次，新建DB需要再次执行。

    --创建
    CREATE extension hg_binlog;
    
    --删除
    DROP extension hg_binlog;

    重要

    不推荐使用DROP EXTENSION <extension_name> CASCADE;命令级联卸载Extension。CASCADE（级联）删除命令不仅会删除指定扩展本身，还会一并清除扩展数据（例如PostGIS数据、RoaringBitmap数据、Proxima数据、Binlog数据、BSI数据等）以及依赖该扩展的对象（包括元数据、表、视图、Server数据等）。

  • Hologres从 V2.0版本起，无需手动创建extension即可使用。

• Hologres V2.1版本起，支持通过如下两种方式进行Binlog消费。

  • 全部版本支持：完成准备工作，包括为目标表创建Publication、为Publication创建Replication Slot后，直接进行目标表的Binlog消费。

    说明

    该方法需授予用户如下权限其中之一：

    • 实例的Superuser权限

    • 目标表的Owner权限、CREATE DATABASE权限及实例的Replication Role权限。

  • 仅Hologres V2.1版本起支持：为用户授予目标表的读权限，然后进行目标表的Binlog消费。

使用限制

• 仅Hologres V1.1及以上版本支持通过JDBC消费Hologres Binlog，如果您的实例是V1.1以下版本，请您使用自助升级或加入Hologres钉钉交流群反馈，详情请参见如何获取更多的在线支持？。

• 仅以下数据类型支持消费Hologres Binlog：INTEGER、BIGINT、SMALLINT、TEXT、CHAR(n)、VARCHAR(n)、REAL、DOUBLE PRECISION、BOOLEAN、NUMERIC(38,8)、DATE、TIME、TIMETZ、TIMESTAMP、TIMESTAMPTZ、BYTEA、JSON、SERIAL、OID、int4[]、int8[]、float4[]、float8[]、boolean[]、text[]，从HologresV1.3.36版本开始支持JSONB类型。如果表中有以上类型之外的数据类型，会造成消费失败。

  说明

  Hologres从V1.3.36版本开始支持JSONB数据类型消费Hologres Binlog，消费之前需要开启如下GUC参数：

  -- Session级别开启GUC
  SET hg_experimental_enable_binlog_jsonb = ON;
  
  -- DB级别开启GUC
  ALTER database <db_name> SET hg_experimental_enable_binlog_jsonb = ON;

• 同普通连接类似，使用JDBC进行Binlog的消费时，所消费的每张表的每个Shard都会使用1个Walsender连接，Walsender连接与普通连接独立，互不影响。

• Walsenders数也有使用上限，可以通过以下命令查看单个Frontend节点的最大Walsender数（V2.2版本起默认值调整为600，V2.0和V2.1版本默认为1000，V1.1.26至V2.0版本之间默认为100），总的数目需要乘实例的Frontend节点数，不同规格实例的Frontend节点数请参见实例规格概述。

  SHOW max_wal_senders;

  说明

  Hologres实例支持的同时消费Binlog的表数量可以通过以下方式计算：表数量<=(max_wal_senders(100或1000) * FrontEnd节点数)/表的Shard Count 。

  例如：

  • 表a和表b的Shard Count都为20，表c的Shard Count为30，则三张表同时进行消费Binlog占用的Walsenders数量为20 + 20 + 30 = 70。

  • 表a和表b的Shard Count都为20，表a同时有两个作业在进行Binlog消费，同时进行消费占用的Walsenders数量为20 *2 + 20 = 60。

  • 一个实例有两个Frontend节点，则其最大Walsenders数为100*2 = 200，最多支持同时消费10张Shard Count为20的表同时进行消费Binlog。

  如果使用JDBC进行Binlog消费的连接数达到上限，会提示FATAL: sorry, too many wal senders already的错误信息，可以按照如下思路进行排查处理：

  1. 检查使用JDBC进行Binlog消费的作业，减少其中非必要的Binlog消费。

  2. 检查Table Group与Shard数设计是否合理，详情请参见Table Group设置最佳实践。

  3. 如连接数仍超出限制，则须考虑扩容实例。

• Hologres V2.0.18版本前，只读从实例不支持通过JDBC消费Binlog功能。从V2.0.18版本起，只读从实例支持通过JDBC消费Hologres Binlog，但不支持记录消费进度。

注意事项

Hologres实例版本和Flink引擎版本不同支持消费Binlog的方式也不同，说明如下：

准备工作：创建Publication和Replication Slot

Hologres V2.1版本前，需要先为目标表创建Publication、为Publication创建Replication Slot后，才可以进行Binlog消费。

Hologres V2.1版本起，除上述方法外，还支持仅有目标表读权限的用户进行Binlog消费。该方法无法查询Hologres侧记录的Binlog消费进度，建议消费端自行记录消费进度。

-- 使用superuser将普通用户设置为replication role：
ALTER role <user_name> replication;

-- 使用superuser将replication role设置回普通用户：
ALTER role <user_name> noreplication;

使用JDBC消费Binlog

1. 添加POM依赖

   使用如下语句添加POM依赖。

   说明

   添加POM依赖，请使用42.2.18及以上版本的JDBC。

           <dependency>
               <groupId>org.postgresql</groupId>
               <artifactId>postgresql</artifactId>
               <version>42.3.8</version>
           </dependency>
           <!-- 用于获取表schema以及解析binlog -->
           <dependency>
               <groupId>com.alibaba.hologres</groupId>
               <artifactId>holo-client</artifactId>
               <version>2.2.10</version>
   				</dependency>

2. Java代码示例

   import com.alibaba.hologres.client.HoloClient;
   import com.alibaba.hologres.client.HoloConfig;
   import com.alibaba.hologres.client.impl.binlog.HoloBinlogDecoder;
   import com.alibaba.hologres.client.model.Record;
   import com.alibaba.hologres.client.model.TableSchema;
   
   import org.postgresql.PGConnection;
   import org.postgresql.PGProperty;
   import org.postgresql.replication.LogSequenceNumber;
   import org.postgresql.replication.PGReplicationStream;
   
   import java.nio.ByteBuffer;
   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.util.Arrays;
   import java.util.List;
   import java.util.Properties;
   
   public class Test {
   
       public static void main (String[] args) throws Exception {
   
           String username = "";
           String password = "";
           String url = "jdbc:postgresql://Endpoint:Port/db_test";
   
           // 创建JDBC连接
           Properties properties = new Properties();
           PGProperty.USER.set(properties, username);
           PGProperty.PASSWORD.set(properties, password);
           PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
           // 消费Binlog，务必加上以下参数
           PGProperty.REPLICATION.set(properties, "database");
           try (Connection connection = DriverManager.getConnection(url, properties)) {
               // 创建PGReplicationStream并绑定Replicaiton slot，需要指定shardId
               int shardId = 0;
               PGConnection pgConnection = connection.unwrap(PGConnection.class);
               PGReplicationStream pgReplicationStream = pgConnection.getReplicationAPI().replicationStream()
                       .logical()
           	  // 2.1版本起，此处有两种可行的写法
               // 方法1：withSlotName参数为准备阶段创建的Replication Slot名，不需填写withSlotOption("table_name","xxx") 
               // 方法2：不需填写withSlotName参数，需填写withSlotOption("table_name","xxx") 
                       .withSlotName("slot_name")
                       .withSlotOption("table_name","public.test_messsage_src") // 消费的表名
                       .withSlotOption("parallel_index", shardId)
                       .withSlotOption("batch_size", "1024")
                       .withSlotOption("start_time", "2021-01-01 00:00:00")
                       .withSlotOption("start_lsn","0")
                       .start();
   	
               // 尽管我们不直接使用holo-client的接口消费binlog，但是需要holo-client的接口去解析消费到的数据。        
               // 创建holo-client
               HoloConfig holoConfig = new HoloConfig();
               holoConfig.setJdbcUrl(url);
               holoConfig.setUsername(username);
               holoConfig.setPassword(password);
               HoloClient client = new HoloClient(holoConfig);
   
               // 创建Binlog decoder用于Decode binary数据，schema需要通过HoloClient获取
               TableSchema schema = client.getTableSchema("test_message_src", true);
               HoloBinlogDecoder decoder = new HoloBinlogDecoder(schema);
   
               // 用于记录当前消费位点，用以在消费中断后，用该值进行继续消费
               Long currentLsn = 0;
               // 消费数据
               ByteBuffer byteBuffer = pgReplicationStream.readPending();
               while (true) {
                   if (byteBuffer != null) {
                       List<BinlogRecord> records = decoder.decode(shardId, byteBuffer);
                       Long latestLsn = 0L;
                       for (BinlogRecord record : records) {
                           latestLsn = record.getBinlogLsn();
                           // Do Something
                           System.out.println( "lsn: " + latestLsn + ", record: " + Arrays.toString(record.getValues()));
                       }
                       // 保存消费位点
                       currentLsn = latestLsn;                    
                       pgReplicationStream.forceUpdateStatus();
                   }
                   byteBuffer = pgReplicationStream.readPending();
               }
           }
           // pgReplicationStream.close();
           // connection.close();
       }

   创建PGReplicationStream时，需要通过withSlotName指定Replication Slot：

   • Hologres V2.1版本前，需要填写已创建的Replication Slot名称。

   • Hologres V2.1版本起，无需填写withSlotName，只需在Slot Options中指定目标表名。

   此外，withSlotOption可以指定如下参数。

   参数	是否必须	说明
   table_name	当不指定withSlotName时必须，否则为无效参数。	当不指定withSlotName时，table_name代表了想要消费的目标表名。格式为schema_name.table_name或table_name。
   parallel_index	是	在使用PGReplicationStream进行Binlog消费时，一个PGReplicationStream会建立1个Walsender连接，对目标表的1个Shard的Binlog进行消费。parallel_index即代表了消费目标表的第parallel_index个Shard的数据。 假设某个表有3个Shard，则通过Replication Slot消费Binlog所需要的并发数为3，则用户最多可以创建3个PGReplicationStream，每个PGReplicationStream的parallel_index参数分别是0、1、2。 当前JDBC消费Hologres Binlog并不支持类似Kafka Consumer Group的实现，所以需要用户自行创建多个PGReplicationStream。
   start_time	否	表示从某个时间点位开始消费Binlog，示例参数格式为：2021-01-01 12:00:00+08。 如果未指定start_lsn或start_time，分为如下三种情况： 第一次开始消费Replication Slot的Binlog，则从头开始消费，类似Kafka的Oldest。 曾经消费过Replication Slot的Binlog，则尝试从之前Commit过的点位开始消费。 对于不指定withSlotName但指定了table_name的使用场景，不论是否曾经消费过该表的Binlog，都会从头开始消费。
   start_lsn	否	表示从某个lsn之后开始消费Binlog，同时设置优先级高于start_time。
   batch_size	否	单次获取的Binlog最大批大小，单位为行，默认值为1024。

   说明

   • BinlogRecord 是decoder返回的Record类型，可以通过以下接口获取这条数据对应的binlog系统字段，详情见订阅Hologres Binlog。

     • getBinlogLsn() 获取binlog的序号。

     • getBinlogTimestamp() 获取Binlog的系统时间戳。

     • getBinlogEventType() 获取Binlog的事件类型。

   • 消费Binlog之后，用户需要手动Commit点位信息，确保下次Failover能够恢复。

使用Holo Client消费Binlog

• 消费Hologres Binlog功能已经集成至Holo Client中，您可以通过指定需要消费的物理表，方便地消费所有Shard的Binlog数据。

• 使用Holo Client消费Binlog，需要占用与物理表shard数（slot并发数）相同的连接数，请保证连接数充足。

• 使用Holo Client消费Binlog过程中，推荐您根据Shard自行保存消费点位，在由于网络连接失败等原因导致消费终止时，可以通过消费点位进行恢复，详情请参见下方代码示例。

1. 添加POM依赖

   使用如下语句添加POM依赖。

   说明

   添加POM依赖，推荐使用2.2.10及以上版本的Holo Client，2.2.9及之前的版本存在内存泄露的问题。

   <dependency>
     <groupId>com.alibaba.hologres</groupId>
     <artifactId>holo-client</artifactId>
     <version>2.2.10</version>
   </dependency>

2. Java代码示例

   import com.alibaba.hologres.client.BinlogShardGroupReader;
   import com.alibaba.hologres.client.Command;
   import com.alibaba.hologres.client.HoloClient;
   import com.alibaba.hologres.client.HoloConfig;
   import com.alibaba.hologres.client.Subscribe;
   import com.alibaba.hologres.client.exception.HoloClientException;
   import com.alibaba.hologres.client.impl.binlog.BinlogOffset;
   import com.alibaba.hologres.client.model.binlog.BinlogHeartBeatRecord;
   import com.alibaba.hologres.client.model.binlog.BinlogRecord;
   
   import java.util.HashMap;
   import java.util.Map;
   
   public class HoloBinlogExample {
   
       public static BinlogShardGroupReader reader;
   
       public static void main(String[] args) throws Exception {
           String username = "";
           String password = "";
           String url = "jdbc:postgresql://ip:port/database";
           String tableName = "test_message_src";
           String slotName = "hg_replication_slot_1";
   
           // 创建client的参数
           HoloConfig holoConfig = new HoloConfig();
           holoConfig.setJdbcUrl(url);
           holoConfig.setUsername(username);
           holoConfig.setPassword(password);
           holoConfig.setBinlogReadBatchSize(128);
           holoConfig.setBinlogIgnoreDelete(true);
           holoConfig.setBinlogIgnoreBeforeUpdate(true);
           holoConfig.setBinlogHeartBeatIntervalMs(5000L);
           HoloClient client = new HoloClient(holoConfig);
   
           // 获取表的shard数
           int shardCount = Command.getShardCount(client, client.getTableSchema(tableName));
   
           // 使用map保存每个shard的消费进度, 初始化为0
           Map<Integer, Long> shardIdToLsn = new HashMap<>(shardCount);
           for (int i = 0; i < shardCount; i++) {
               shardIdToLsn.put(i, 0L);
           }
   
           // 消费binlog的请求，2.1版本前tableName和slotname为必要参数，2.1版本起仅需传入tableName（等价于前文使用的固定slotName“hg_table_name_slot”）。
           // Subscribe有StartTimeBuilder和OffsetBuilder两种，此处以前者为例
           Subscribe subscribe = Subscribe.newStartTimeBuilder(tableName, slotName)
                   .setBinlogReadStartTime("2021-01-01 12:00:00")
                   .build();
           // 创建binlog reader
           reader = client.binlogSubscribe(subscribe);
   
           BinlogRecord record;
   
           int retryCount = 0;
           long count = 0;
           while(true) {
               try {
                   if (reader.isCanceled()) {
                       // 根据保存的消费点位重新创建reader
                       reader = client.binlogSubscribe(subscribe);
                   }
                   while ((record = reader.getBinlogRecord()) != null) {
                       // 消费到最新
                       if (record instanceof BinlogHeartBeatRecord) {
                           // do something
                           continue;
                       }
       
                       // 处理读取到的binlog record，这里只做打印
                       System.out.println(record);
       
                       // 处理之后保存消费点位，异常时可以从此点位恢复
                       shardIdToLsn.put(record.getShardId(), record.getBinlogLsn());
                       count++;
                       
                       // 读取成功，重置重试次数
                       retryCount = 0;
                   }
               } catch (HoloClientException e) {
                   if (++retryCount > 10) {
                       throw new RuntimeException(e);
                   }
                   // 发生异常时推荐打印warn级别日志
                   System.out.println(String.format("binlog read failed because %s and retry %s times", e.getMessage(), retryCount));
       
                   // 重试期间进行一定时间的等待
                   Thread.sleep(5000L * retryCount);
       
                   // 用OffsetBuilder创建Subscribe，从而为每个shard指定起始消费点位
                   Subscribe.OffsetBuilder subscribeBuilder = Subscribe.newOffsetBuilder(tableName, slotName);
                   for (int i = 0; i < shardCount; i++) {
                       // BinlogOffset通过setSequence指定lsn，通过setTimestamp指定时间，两者同时指定lsn优先级大于时间戳
                       // 这里根据shardIdToLsn这个Map中保存的消费进度进行恢复
                       subscribeBuilder.addShardStartOffset(i, new BinlogOffset().setSequence(shardIdToLsn.get(i)));
                   }
                   subscribe = subscribeBuilder.build();
                   // 关闭reader
                   reader.cancel();
               }
           }
       }
   }

   使用Holo Client消费Binlog时可以指定如下参数。

   参数	是否必须	默认值	说明
   binlogReadBatchSize	否	1024	从每个Shard单次获取的Binlog最大批次大小，单位为行。
   binlogHeartBeatIntervalMs	否	-1	binlogRead发送BinlogHeartBeatRecord的间隔，-1表示不发送。 当Binlog没有新数据，每间隔binlogHeartBeatIntervalMs会下发一条BinlogHeartBeatRecord，此record的timestamp表示截止到这个时间这个Shard上的数据都已经消费完成。
   binlogIgnoreDelete	否	false	是否忽略Delete类型的Binlog。
   binlogIgnoreBeforeUpdate	否	false	是否忽略BeforeUpdate类型的Binlog。

常见问题

消费Binlog并提交消费进度后，发现表hologres.hg_replication_progress不存在，或表中没有消费进度数据，可能原因如下：

• 消费时不通过Replication Slot进行，即不指定参数withSlotName，该场景不支持记录消费进度。

• 使用了只读从实例，且该DB是第一次被消费Binlog，此时hologres.hg_replication_progress表创建失败。Hologres V2.0.18版本起已修复，从实例可以正常消费Binlog。Hologres V2.0.18版本前，需要使用主实例先消费一次Binlog，从实例即可正常消费。

• 如果不是上述原因，请加入Hologres钉钉交流群联系值班人员处理，详情请参见如何获取更多的在线支持？。