Flink / Scala - 使用 Jedis、JedisPool 作为 Source 读取数据

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
云原生内存数据库 Tair,内存型 2GB
云数据库 Redis 版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 现在有一批数据写入多台 Redis 相同 key 的队列中,需要消费 Redis 队列作为 Flink Source,为了提高可用性,下面基于 JedisPool 进行队列的消费。

一.引言

现在有一批数据写入多台 Redis 相同 key 的队列中,需要消费 Redis 队列作为 Flink Source,为了提高可用性,下面基于 JedisPool 进行队列的消费。队列数据示例: 1,2,3,4,5、A,B,C,D,E,程序将字符串解析并 split(",") 然后分别写到下游。

二.Flink Source By JedisPool

1.初始化 JedisPool

由于数据量较大,所以同时写入 N 台 Redis 队列,key 均相同,注意这里是 JedisPool 不是 JedisCluster,需要区分二者的概念。

def initJedisPool(host: String, port: Int): JedisPool = {
    val config = new JedisPoolConfig
    config.setMaxTotal(4)
    config.setMaxIdle(2)
    config.setMaxWaitMillis(1000)
    config.setTestOnBorrow(true)
    config.setTestOnReturn(true)
    jedisPool = new JedisPool(config, host, port)
    jedisPool
  }

image.gif

需要导入 Jedis 依赖:

<dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.7.2</version>
        </dependency>

image.gif

这里 JedisPool 的几个参数可以参考之前的文章:JedisPool - Java.net.SocketException: Broken pipe (write failed),里面介绍了 TestOnBorrow、TestOnReturn 的含义。

2.实现 RedisSource

自定义 Source 需要继承 org.apache.flink.streaming.api.functions.source.RichSourceFunction 实现 run 方法和 cancel 方法即可,主要逻辑在 run 方法中,run 方法负责从队列中不断获取数据并 collect 到下游,除此之外,由于需要读取 redis client,所以还需要新增 open 方法进行 client 的初始化。

class JedisPoolSourceTest(host: String, port: Int) extends RichSourceFunction[String] {
  var jedisPool: JedisPool = _ // JedisPool
  val listKey = "testListKey" // 公用队列 Key
  override def open(parameters: Configuration): Unit = {
    initJedisPool(host, port) // 初始化 JedisPool
  }
  // 消费 Redis 队列产出数据
  override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
    collectData(sourceContext)
  }
  // 关闭 Source
  override def cancel(): Unit = {
    jedisPool.close()
  }
}

image.gif

3.产出数据 CollectData

上面是继承 RichSourceFunction 完成了整体的架构,下面单独把 collectData 的数据产出逻辑分析一下:

def collectData(out: SourceFunction.SourceContext[String]): Unit = {
    var resource: Jedis = null
    try {
      resource = jedisPool.getResource // 获取 Jedis 实例
      // while True 保证持续消费
      while (true) {
        // 解析并产出
        val info = resource.lpop(listKey)
        if (info != null) {
          val sendRe = info.split(",")
          if (sendRe.nonEmpty) {
            sendRe.foreach(out.collect)
          }
        }
        // 如果队列为空则 Sleep 1s
        if (resource.llen(listKey) == 0) {
          TimeUnit.SECONDS.sleep(1)
        }
      }
    } catch {
      case e: Exception => {
        e.printStackTrace()
        resource.close() // 关闭资源
        collectData(out) // 出现异常递归重启,保证 Source 的可用性
      }
    }
  }

image.gif

collectData 为 run 方法的主类,主要用于消费队列并产出数据到下游,有几个点需要注意:

A.获取 Jedis 实例

由于本例中使用 JedisPool,所以这里采用 pool.getResource 的方式获取 Jedis,网上也有一些 demo 在这里直接初始化单独 JedisClient,这里采用 pool 的形式主要是考虑到稳定性的情况。

B.While True

这里没啥太多说的,通过 while true 实现不间断的消费队列数据。

C.Sleep(1)

这里考虑到队列为空时,如果频繁 while true 访问 Jedis 会造成高 QPS 且无用操作,所以这里检测到队列为空时会加入 1s 的延迟,这个时间也可以根据自己的任务场景灵活修改。

D.Try Catch + 递归

try-catch 逻辑内除关闭 resource 外还递归调用了 CollectData,该方法参考了 Spark-Streaming 的 receiver 方法,在故障时能够重新申请 redis 连接并重新读取数据,保证 Source 的稳定性,Spark-Streaming 的 receiver 可以参考:Spark Streaming Receiver restart 重启

4.合并多条数据源

由于数据量较大,所以数据分布式的写入到多台 Redis 队列中,上面已经实现了 JedisSource,下面则需要将多台 Redis 分别接入 JedisSource 并绑定合成一个统一的 Source。

var dataStream: DataStream[String] = null // 初始化原始流
    // 遍历多台 Redis 的 Host && Port
    testHostAndPorts.foreach { case (host, port) => { 
      if (dataStream == null) {
        dataStream = env.addSource[String](new JedisPoolSourceTest(host, port))
      } else {
        val newStream = env.addSource[String](new JedisPoolSourceTest(host, port))
        dataStream = dataStream.union(newStream) // union 合并流
      }
    }
    }

image.gif

通过 var 构造可变变量,然后不断将 RedisSource union 到 DataStream 中,最终形成统一流。

三.测试多 Redis Source

1.启动多台 client

启动多个 Jedis 连接并向队列 key 推数据:

redis-cli -h $host -p $port

image.gif

在不同客户端分别执行:

lpush testListKey 0,1,2,3,4
lpush testListKey A,B,C,D,E

image.gif

2.主函数 main

主函数逻辑比较简单,主要就是接受队列数据,"," split 并将每个元素输出,最终 print sink 到标准输出。

def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment // 执行环境
    // 合并多流
    var dataStream: DataStream[String] = null
    testHostAndPorts.foreach { case (host, port) => {
      if (dataStream == null) {
        dataStream = env.addSource[String](new JedisPoolSourceTest(host, port))
      } else {
        val newStream = env.addSource[String](new JedisPoolSourceTest(host, port))
        dataStream = dataStream.union(newStream)
      }
    }
    }
    // 标准输出
    dataStream.print()
    // 执行
    env.execute()

image.gif

上面通过 2 台 client 分别 lpush 了数据,这边接收测试也没有问题:

image.gif编辑

四.尝试与优化

1.多线程的尝试

上面采用 JedisPool + 单 Resource 的形式进行接受数据,其实最早使用 JedisPool 是为了接入多线程 Source,即在 run 方法内使用线程池提高生产效率:

def addTask(poolSize: Int, jedisPool: JedisPool, sourceContext: SourceFunction.SourceContext[String]): Unit = {
    val executor = Executors.newFixedThreadPool(2) // 初始化线程池
    (0 until poolSize).foreach(epoch => {
      executor.submit(new Runnable {
        override def run(): Unit = {
          val resource: Jedis = jedisPool.getResource
          while (true) {
            val info = resource.lpop(listKey)
            if (info != null) {
              info.split(",").foreach(sourceContext.collect)
            }
            if (resource.llen(listKey) == 0) {
              TimeUnit.SECONDS.sleep(1)
            }
          }
        }
      })
    })
  }

image.gif

这里 Runnable 和上面单 Resource 接队列消费逻辑相同,实现后发现可以多线程读取到 Jedis 队列的内容,随后程序发生堵塞,经过调试发现任务卡在 sourceContext.collect,不知道是不是 sourceContext 不支持多线程,有了解的同学也可以评论区讨论一下~

2.空闲数据源的优化

上述多台 redis 队列并不能保证全天无间断都有数据产出,在 EventTime 场景下,会出现流数据空闲的状态从而影响 WaterMark 的更新影响整个任务的时间推进,为了解决这个问题,可以使用新版的时间戳策略即 WatermarkStrategy:

WatermarkStrategy
      .forBoundedOutOfOrderness[T](Duration.ofSeconds(60))
      .withIdleness(Duration.ofMinutes(1))

image.gif

通过 org.apache.flink.api.common.eventtime.WatermarkStrategy 的 withIdleness 方法对数据流进行标记,当数据源在 Duration 规定的时间内未产出数据时将该 Source 标记为空闲状态,这样下游的数据也不需要等待该 Source 的 WaterMark 从而保证数据流的正常推进,待有新数据推入 Source 中,该数据流切换为活跃状态,重新向下游发送其真实 WaterMark 水印。

3.数据量过大

对于数据过多或过大的场景,lpop 可能会有一定的读取和网络压力,这时候可以采用 lrange + ltrim 的批量读取逻辑,将单条访问改为批量访问,减少 redis 读取压力。其思想与 hscan 代替 hgetAll 批量访问过大数据有一定类似,都是采取分治的思想,只不过前者是由一到多,后者是化整为零。

五.总结

绑定多台 Redis 源上线后,任务没有问题且支持空流处理,除了 Redis Source 外,还有 Redis Sink 相关的实现,大家可以参考: Flink / Scala - 使用 RedisSink 存储数据,这里使用 SharedJedisPool 代替了 Flink 自带的 RedisCommandsContainer,后续也会单独出一期 RedisCommandsContainer 的 Flink-Jedis 教程。

image.gif编辑

可以看到多台 Redis Source 上线后,这个作业图实在是 🚄。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
13天前
|
SQL 安全 数据处理
揭秘数据脱敏神器:Flink SQL的神秘力量,守护你的数据宝藏!
【9月更文挑战第7天】在大数据时代,数据管理和处理尤为重要,尤其在保障数据安全与隐私方面。本文探讨如何利用Flink SQL实现数据脱敏,为实时数据处理提供有效的隐私保护方案。数据脱敏涉及在处理、存储或传输前对敏感数据进行加密、遮蔽或替换,以遵守数据保护法规(如GDPR)。Flink SQL通过内置函数和表达式支持这一过程。
37 2
|
25天前
|
调度 流计算
Flink 新一代流计算和容错问题之Flink 中的数据可以分为什么类型
Flink 新一代流计算和容错问题之Flink 中的数据可以分为什么类型
|
29天前
|
SQL 数据库 流计算
Flink CDC数据读取问题之一致性如何解决
Flink CDC 使用Change Data Capture (CDC)技术从数据库捕获变更事件,并利用Flink的流处理能力确保数据读取一致性。相较于传统工具,它具备全增量一体化数据集成能力,满足实时性需求。在实践中解决了高效数据同步、稳定同步大量表数据等问题。应用场景包括实时数据同步、实时数据集成等。快速上手需学习基本概念与实践操作。未来发展方向包括提升效率与稳定性,并依据用户需求持续优化。
70 1
|
22天前
|
消息中间件 Kafka 数据处理
实时计算 Flink版产品使用问题之如何处理数据并记录每条数据的变更
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
22天前
|
消息中间件 存储 关系型数据库
实时计算 Flink版产品使用问题之同步时,上游批量删除大量数据(如20万条),如何提高删除效率
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
22天前
|
消息中间件 存储 关系型数据库
实时计算 Flink版产品使用问题之如何使用Kafka Connector将数据写入到Kafka
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
22天前
|
SQL 关系型数据库 HIVE
实时计算 Flink版产品使用问题之如何将PostgreSQL数据实时入库Hive并实现断点续传
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
22天前
|
消息中间件 监控 Kafka
实时计算 Flink版产品使用问题之处理Kafka数据顺序时,怎么确保事件的顺序性
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
22天前
|
SQL 分布式计算 数据处理
实时计算 Flink版产品使用问题之怎么将数据从Hive表中读取并写入到另一个Hive表中
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。