从Redis读取数据到Flink

Question

我一直在试图找到一个连接器，将数据从Redis读取到Flink。Flink的文档中包含了要写入Redis的连接器的描述。在我的Flink工作中，我需要从Redis读取数据。在使用ApacheFlink进行数据流传输时，Fabian提到可以从Redis读取数据。可用于此目的的接头是什么？

Answer 1

关于为ApacheFlink提供流式redis源代码连接器（请参阅Flink-3033），已经有过一些讨论，但目前还没有可用的。然而，实现一个应该并不困难。

Answer 2

我们正在生产中运行一个大致看起来像这样的

class RedisSource extends RichSourceFunction[SomeDataType] {

  var client: RedisClient = _

  override def open(parameters: Configuration) = {
    client = RedisClient() // init connection etc
  }

  @volatile var isRunning = true

  override def cancel(): Unit = {
    isRunning = false
    client.close()
  }

  override def run(ctx: SourceContext[SomeDataType]): Unit = while (isRunning) {
      for {
        data <- ??? // get some data from the redis client
      } yield ctx.collect(SomeDataType(data))

  }
} 

我认为这真的取决于你需要从redis获取什么。上面的方法可以用来从列表/队列中获取消息，转换/推送，然后从队列中删除它。Redis还支持Pub/Sub，因此可以订阅、获取SourceConext并向下游推送消息。

Answer 3

目前，Flink Redis连接器不可用，但可以通过扩展RichSinkFunction/SinkFunction类来实现。

public class RedisSink extends RichSinkFunction<String> {

  @Override
  public void open(Configuration parameters) throws Exception {
      //open redis connection
  }

  @Override
  public void invoke(String map) throws Exception {
     //sink data to redis
  }

  @Override
  public void close() throws Exception {
     super.close();
  }

}