如何查询flink的可查询状态

Question

我正在使用 flink 1.8.0，我正在尝试查询我的作业状态。

val descriptor = new ValueStateDescriptor("myState", Types.CASE_CLASS[Foo])
    descriptor.setQueryable("my-queryable-State")

我使用端口 9067，根据 this，这是默认端口，我的客户：

val client = new QueryableStateClient("127.0.0.1", 9067)
val jobId = JobID.fromHexString("d48a6c980d1a147e0622565700158d9e")

      val execConfig = new ExecutionConfig
       val descriptor = new ValueStateDescriptor("my-queryable-State", Types.CASE_CLASS[Foo])
      val res: Future[ValueState[Foo]] = client.getKvState(jobId, "my-queryable-State","a", BasicTypeInfo.STRING_TYPE_INFO, descriptor)
      res.map(_.toString).pipeTo(sender)

但我得到了：

[ERROR] [06/25/2019 20:37:05.499] [bvAkkaHttpServer-akka.actor.default-dispatcher-5] [akka.actor.ActorSystemImpl(bvAkkaHttpServer)] Error during processing of request: 'org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /127.0.0.1:9067'. Completing with 500 Internal Server Error response. To change default exception handling behavior, provide a custom ExceptionHandler.
java.util.concurrent.CompletionException: org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /127.0.0.1:9067

我做错了什么？
我应该如何以及在哪里定义 QueryableStateOptions

Answer 1

所以如果你想使用 QueryableState 你需要添加合适的 Jar 到你的 flink。该 jar 是 flink-queryable-state-runtime，它可以在您的 flink 发行版的 opt 文件夹中找到，您应该将它移动到 lib 文件夹。

至于第二个问题，QueryableStateOption 只是一个用于创建静态 ConfigOption 定义的 class。然后使用这些定义从 flink-conf.yaml 文件中读取配置。所以目前配置 QueryableState 的唯一选择是使用 flink 分发中的 flink-conf 文件。

编辑：另外，请尝试阅读这篇文章]1，它提供了有关可查询状态如何工作的更多信息。您不应该真正直接连接到服务器端口，而应该使用默认为 9069.

的代理端口

如何查询flink的可查询状态

How to query flink's queryable state

scala

apache-flink

flink-sql