使用 Flink 从 Redis 读取数据答案

【问题标题】：Read Data From Redis Using Flink使用 Flink 从 Redis 读取数据
【发布时间】：2018-11-21 21:30:19
【问题描述】：

我对 Flink 完全陌生。可能会重复这个问题，但只找到一个链接，这对我来说是无法理解的。

https://stackoverflow.com/a/44294980/6904987

我在 Redis 中以 Key Value 格式存储数据，例如 Key 是 UserId，UserInfo 是 value。写在下面的代码。

class RedisExampleMapper extends RedisMapper[(String, String)] {
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "HASH_NAME")
  }

  override def getKeyFromData(data: (String, String)): String = data._1

  override def getValueFromData(data: (String, String)): String = data._2
}


val env = StreamExecutionEnvironment.getExecutionEnvironment
    val conf = new FlinkJedisPoolConfig.Builder().setHost("IP").build()
    val streamSink = env.readTextFile("/path/useInformation.txt").map(x => {
          val userInformation = x.split(",")
          val UserId = userInformation(0)
          val UserInfo  = userInformation(1)
          (UserId , UserInfo)
        })
val redisSink = new RedisSink[(String, String)](conf, new RedisExampleMapper)
streamSink.addSink(redisSink)

样本数据：

12“用户信息12”

13“用户信息13”

14“用户信息14”

15“用户信息15”

我想使用基于 key 的 Flink 从 redis 获取数据。示例 14 应返回“UserInfo14”。输出应该打印在 Flink 日志文件或终端中，无论它是什么。

提前致谢。

【问题讨论】：

请扩展您想要做什么的描述。你将从 Redis 中读取多少数据？您的问题表明这可能只是一个（或很少）记录。如果这是真的，那么 Flink 不是适合这项工作的工具，因为它增加了太多开销。 Flink 是一个用于并行处理大量数据（流式或批处理）的框架。该框架为仅获取单个值增加了太多开销。

标签： scala apache-flink flink-streaming flink-sql

【解决方案1】：

在https://stackoverflow.com/a/44294980/6904987 中扩展答案。

使用env.addSource(new RedisSource(data structure name)) 添加源。

您必须自己实现连接到 Redis 数据库的 RedisSource，从 Redis 数据结构中读取记录。

实施取决于。您可以通过轮询从 Redis 消费，也可以订阅 Redis，只要您从 Redis 获取事件，就从源发出事件。

您可以在此处查看一般 SourceFunction 示例和文档：https://ci.apache.org/projects/flink/flink-docs-release-1.5/api/java/org/apache/flink/streaming/api/functions/source/SourceFunction.html

【讨论】：

【解决方案2】：

如果要查询 Redis 进行键值搜索，可以在转换中使用 Redis 客户端。例如，如果您使用 Java 和 Flink，可以使用 Jedis 查询 Redis。

【讨论】：