scala—如何使用api rest，将flink流作为参数传递并返回已转换的流

vhipe2zx 于 2021-06-24 发布在 Flink

关注(0)|答案(2)|浏览(332)

我是Apache·Flink的新手。我有一个flink scala项目，它使用kafka集群中的数据，我需要将流结果作为参数传递给使用api的用户，该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()
  }   
}

有什么帮助吗？

scala apache-flink flink-streaming scala-streams

来源：https://stackoverflow.com/questions/60129074/how-to-consume-api-rest-passing-flink-stream-as-parameter-and-return-this-stream

2条答案

按热度按时间

gmol16391#

这是我最后的密码。我希望这有帮助

class Testing extends Serializable{
  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)
    // Here I need pass stream as parameter to api and return transformed stream
    val result = stream.flatMap{
      (str, out: Collector[String]) =>
        val api_test = "http://api-test.server.local/test/?msg=%s"
        out.collect {
          getUrl(api_test.format(URLEncoder.encode(str, "UTF-8")))
        }        
    }    
    env.execute()
  }

  def getUrl(url: String): String = {
    val timeout = 5
    val config = RequestConfig.custom.setConnectTimeout(timeout * 1000).setConnectionRequestTimeout(timeout * 1000).setSocketTimeout(timeout * 1000).build
    val client: CloseableHttpClient = HttpClientBuilder.create.setDefaultRequestConfig(config).build
    val request = new HttpGet(url)
    val response = client.execute(request)
    val entity = response.getEntity
    val get_result = EntityUtils.toString(entity)
    get_result
  }     
}

赞(0）回复(0）举报 2021-06-24

jobtbby32#

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

赞(0）回复(0）举报 2021-06-24