Apache Beam RedisIO如何使用PFCOUNT获取HyperLogLog唯一值计数

方案1：扩展官方RedisIO连接器（长期最优方案） #

你定位的确实是Beam现有RedisIO的实现缺口，目前官方连接器只封装了字符串、哈希、列表等常用结构的操作，没有覆盖HyperLogLog相关命令。你可以直接在现有RedisIO的命令封装层新增PFCOUNT的适配逻辑，参考现有命令的实现规范调整你的函数逻辑即可，示例如下：

// 对齐Beam RedisIO的客户端调用规范的实现示例
public Response<Long> readUsingPFCOUNTCommand(String key) {
  return connection.execute(jedis -> jedis.pfcount(key));
}

修改完成后打包替换你项目里的Beam Redis相关依赖即可，也可以提交PR到官方社区，后续版本会合并这个能力。

方案2：自定义ParDo调用原生Redis客户端（快速适配方案） #

如果不想修改官方依赖代码，可以自己实现一个处理函数，直接调用Redis原生客户端执行PFCOUNT，不需要改动现有依赖，适合快速上线的场景：

public class PFCountProcessor extends DoFn<String, Long> {
  private transient JedisPool jedisPool;
  private final String redisHost;
  private final int redisPort;

  public PFCountProcessor(String redisHost, int redisPort) {
    this.redisHost = redisHost;
    this.redisPort = redisPort;
  }

  @Setup
  public void initConnection() {
    jedisPool = new JedisPool(redisHost, redisPort);
  }

  @ProcessElement
  public void count(@Element String redisKey, OutputReceiver<Long> output) {
    try (Jedis jedis = jedisPool.getResource()) {
      output.output(jedis.pfcount(redisKey));
    }
  }

  @Teardown
  public void closeConnection() {
    if (jedisPool != null) {
      jedisPool.close();
    }
  }
}

使用时直接把你的Redis key输入流应用这个ParDo即可：

PCollection<String> redisKeys = ...; // 你的key输入流
PCollection<Long> pfCountResults = redisKeys.apply(ParDo.of(new PFCountProcessor("redis-host", 6379)));

注意事项 #

• 自定义ParDo时必须用连接池管理Redis连接，禁止每次请求新建连接，避免Redis实例连接数溢出
• PFCOUNT支持传入多个key返回合并后的去重基数，如果有批量统计需求可以调整入参逻辑，减少Redis请求次数
• 如果你用的是Redis集群模式，需要注意多个key如果不在同一个哈希槽的话，原生PFCOUNT会报错，需要提前做key哈希槽路由处理

内容的提问来源于stack exchange，提问作者Roberto Santos