如何使用api rest通过flink流作为参数并返回转换后的流

Posted 2023-01-25

我是Apache Flink的新手。我有一个flink scala项目，它消耗来自kafka集群的数据，我需要将流结果作为参数传递，以消耗返回此已转换流的api。这是我的代码

class Testing 
  def main(args: Array[String]): Unit = 
  def streamTest(): Unit = 
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "test1.server.local:9092,test2.server.local:9092,test3.server.local:9092")
    val consumer_test = new FlinkKafkaConsumer[String]("topic_test", new SimpleStringSchema(), properties)
    consumer_test.setStartFromEarliest()
    val stream =  env.addSource(consumer_test).setParallelism(5)
    val api_test = "http://api-test.server.local/test/?msg=%s"
    // Here I need pass stream as parameter to api and return transformed stream
    env.execute()
     

是否有帮助？

答案

您应该使用您熟悉的任何http / rest库，然后使用asyncIO。