可查询状态
— 焉知非鱼Queryable State Beta
注意:可查询状态的客户端 API 目前处于不断发展的状态,对所提供接口的稳定性不做保证。在即将到来的 Flink 版本中,客户端的 API 很可能会有突破性的变化。
简而言之,这个功能将 Flink 的 managed keyed (partitioned) state(参见 Working with State)暴露给外界,并允许用户从 Flink 外部查询作业的状态。对于某些场景来说,可查询状态消除了与外部系统(如键值存储)进行分布式操作/交易的需求,而这往往是实践中的瓶颈。此外,该功能对于调试目的可能特别有用。
注意事项: 当查询一个状态对象时,该对象是在没有任何同步或复制的情况下从一个并发线程访问的。这是一个设计上的选择,因为上述任何一种情况都会导致作业延迟的增加,这是我们想要避免的。因为任何使用 Java 堆空间的状态后端,如 MemoryStateBackend 或 FsStateBackend,在检索值时都不会使用副本,而是直接引用存储的值,所以读-修改-写模式是不安全的,可能会导致可查询状态服务器因并发修改而失败。RocksDBStateBackend 则可以避免这些问题。
架构 #
在展示如何使用可查询状态之前,先简单介绍一下构成它的实体。Queryable State 功能由三个主要实体组成。
- QueryableStateClient,它(可能)运行在 Flink 集群之外,并提交用户查询。
- QueryableStateClientProxy,它运行在每个 TaskManager 上(即 Flink 集群内部),负责接收客户端的查询,代表他从负责的 TaskManager 中获取所请求的状态,并将其返回给客户端,以及
- QueryableStateServer,它运行在每个 TaskManager 上,负责为本地存储的状态提供服务。
客户端连接到其中一个代理,并发送一个与特定键 k 相关联的状态的请求。正如在使用状态中所述,keyed state 被组织在键组(Key Groups)中,每个 TaskManager 都被分配了一些这样的键组(Key Groups)。为了发现哪个 TaskManager 负责持有 k 的键组,代理将询问 JobManager。根据答案,代理将查询运行在该 TaskManager 上的 QueryableStateServer,以获取与 k 相关联的状态,并将响应转发到客户端。
激活可查询状态 #
要在 Flink 集群上启用可查询状态,你需要做以下工作。
- 将
flink-queryable-state-runtime_2.11-1.11.0.jar
从 Flink 发行版的opt/
文件夹中复制到lib/
文件夹中。 - 设置属性
queryable-state.enable
为true
。请参阅配置文档了解详情和附加参数。
要验证您的群集是否在启用可查询状态后运行,请检查任何 TaskManager 的日志中的行。“Started the Queryable State Proxy Server @ …"。
使状态可查询 #
现在你已经在集群上激活了可查询状态,现在是时候看看如何使用它了。为了使一个状态对外界可见,它需要通过使用以下方式明确地成为可查询状态。
- QueryableStateStream, 一个方便的对象,它作为一个接收器(sink),并把它的传入值作为可查询的状态提供,或者是
- stateDescriptor.setQueryable(String queryableStateName) 方法,使得状态描述符所代表的 keyed state,可以查询。
下面的章节将解释这两种方法的使用。
可查询的状态流 #
在 KeyedStream 上调用 .asQueryableState(stateName, stateDescriptor)
会返回一个 QueryableStateStream
,它将其值作为可查询状态提供。根据状态的类型,asQueryableState()
方法有以下几种变体。
// ValueState
QueryableStateStream asQueryableState(
String queryableStateName,
ValueStateDescriptor stateDescriptor)
// Shortcut for explicit ValueStateDescriptor variant
QueryableStateStream asQueryableState(String queryableStateName)
// FoldingState
QueryableStateStream asQueryableState(
String queryableStateName,
FoldingStateDescriptor stateDescriptor)
// ReducingState
QueryableStateStream asQueryableState(
String queryableStateName,
ReducingStateDescriptor stateDescriptor)
注意:没有可查询的 ListState
接收器,因为这会导致一个不断增长的列表,可能无法清理,因此最终会消耗过多的内存。
返回的 QueryableStateStream
可以被看作是一个接收器(sink),不能被进一步转换。在内部,一个 QueryableStateStream
会被翻译成一个操作符,它使用所有传入的记录来更新可查询状态实例。更新逻辑是由 asQueryableState
调用中提供的 StateDescriptor
的类型暗示的。在像下面这样的程序中,keyed stream 的所有记录将通过 ValueState.update(value)
来更新状态实例:
stream.keyBy(0).asQueryableState("query-name")
这就像 Scala API 的 flatMapWithState
一样。
管理的 Keyed State #
通过 StateDescriptor.setQueryable(String queryableStateName)
使相应的状态描述符成为可查询的状态,可以使操作符的托管键控状态(参见使用 Managed Keyed State))成为可查询的状态,如下面的例子。
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average", // the state name
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})); // type information
descriptor.setQueryable("query-name"); // queryable state name
注意:queryableStateName
参数可以任意选择,并且只用于查询。它不一定要与状态本身的名称相同。
这个变体对于哪种类型的状态可以被查询没有限制。这意味着它可以用于任何 ValueState、ReduceState、ListState、MapState、AggregatingState 以及目前已被废弃的 FoldingState。
查询状态 #
到目前为止,你已经设置了你的集群以可查询的状态运行,并且你已经将你的(部分)状态声明为可查询。现在是时候看看如何查询这个状态了。
为此,你可以使用 QueryableStateClient
辅助类。它可以在 flink-queryable-state-client jar
中找到,它必须和 flink-core 一起被显式地包含在项目的 pom.xml 中作为依赖,如下所示。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>1.11.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-queryable-state-client-java</artifactId>
<version>1.11.0</version>
</dependency>
更多的内容,可以查看如何设置 Flink 程序。
QueryableStateClient
会将你的查询提交给内部代理,然后代理会处理你的查询并返回最终结果。初始化客户端的唯一要求是提供一个有效的 TaskManager 主机名(记住每个 TaskManager 上都有一个可查询状态代理运行)和代理监听的端口。更多关于如何配置代理和状态服务器端口的信息请参见配置部分。
QueryableStateClient client = new QueryableStateClient(tmHostname, proxyPort)
客户端准备好后,要查询一个类型为 V 的状态,与类型为 K 的键相关联,可以使用该方法。
CompletableFuture<S> getKvState(
JobID jobId,
String queryableStateName,
K key,
TypeInformation<K> keyTypeInfo,
StateDescriptor<S, V> stateDescriptor)
以上返回一个 CompletableFuture,最终持有 ID 为 jobID 的作业的 queryableStateName
所标识的可查询状态实例的状态值。key 是你对其状态感兴趣的键,keyTypeInfo 将告诉 Flink 如何序列化/解序列化它。最后,stateDescriptor
包含了关于所请求的状态的必要信息,即它的类型(Value、Reduce 等)和如何序列化/解序列化它的必要信息。
细心的读者会注意到,返回的 future 包含一个 S 类型的值,即一个包含实际值的 State
对象。这可以是 Flink 支持的任何一种状态类型。ValueState,ReduceState,ListState,MapState,AggregatingState,以及目前已经废弃的 FoldingState。
注意:这些状态对象不允许对包含的状态进行修改。您可以使用它们来获取状态的实际值,例如使用 valueState.get()
,或者迭代包含的 <K,V>
条目,例如使用 mapState.entry()
,但您不能修改它们。举个例子,在返回的列表状态上调用 add()
方法会抛出一个 UnsupportedOperationException
。
注意:客户端是异步的,可以被多个线程共享。在未使用时需要通过 QueryableStateClient.shutdown()
来关闭它,以释放资源。
例子 #
下面的例子扩展了 CountWindowAverage
的例子(请看使用 Managed Keyed State),使其可查询,并展示了如何查询这个值。
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private transient ValueState<Tuple2<Long, Long>> sum; // a tuple containing the count and the sum
@Override
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
Tuple2<Long, Long> currentSum = sum.value();
currentSum.f0 += 1;
currentSum.f1 += input.f1;
sum.update(currentSum);
if (currentSum.f0 >= 2) {
out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
sum.clear();
}
}
@Override
public void open(Configuration config) {
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average", // the state name
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})); // type information
descriptor.setQueryable("query-name");
sum = getRuntimeContext().getState(descriptor);
}
}
Once used in a job, you can retrieve the job ID and then query any key’s current state from this operator:
QueryableStateClient client = new QueryableStateClient(tmHostname, proxyPort);
// the state descriptor of the state to be fetched.
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average",
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}));
CompletableFuture<ValueState<Tuple2<Long, Long>>> resultFuture =
client.getKvState(jobId, "query-name", key, BasicTypeInfo.LONG_TYPE_INFO, descriptor);
// now handle the returned value
resultFuture.thenAccept(response -> {
try {
Tuple2<Long, Long> res = response.get();
} catch (Exception e) {
e.printStackTrace();
}
});
配置 #
以下配置参数会影响可查询状态服务器和客户端的行为,它们被定义在 QueryableStateOptions
中。
状态服务器 #
queryable-state.server.ports
:可查询状态服务器的服务器端口范围。如果在同一台机器上运行多个 task manager,这对避免端口冲突很有用。指定的范围可以是:一个端口: “9123”,一个端口范围: “50100-50200”,或者一个范围和或点的列表: “50100-50200,50300-50400,51234”。默认端口为 9067。queryable-state.server.network-threads
: 接收状态服务器传入请求的网络(事件循环)线程数(0 => #slots)。queryable-state.server.query-threads
: 为状态服务器处理/服务传入请求的线程数(0 => #slots)。
代理 #
queryable-state.proxy.ports
:可查询状态代理服务器的端口范围。如果在同一台机器上运行多个 task manager,这对避免端口冲突很有用。指定的范围可以是:一个端口: “9123”,一个端口范围: “50100-50200”,或者一个范围和或点的列表: “50100-50200,50300-50400,51234”。默认端口为 9069。queryable-state.proxy.network-threads
:为客户端代理接收传入请求的网络(事件循环)线程数(0 => #slots)。queryable-state.proxy.query-threads
:为客户端代理处理/服务传入请求的线程数(0 => #slots)。
限制条件 #
- 可查询状态的生命周期与任务的生命周期绑定,例如,任务在启动时注册可查询状态,在处置时取消注册。在未来的版本中,我们希望将其解耦,以便在任务完成后允许查询,并通过状态复制加快恢复速度。
- 关于可用 KvState 的通知是通过一个简单的告诉发生的。将来应该改进这个功能,使其更加强大,包括询问和确认。
- 服务器和客户端会跟踪查询的统计数据。目前默认情况下,这些数据是被禁用的,因为它们不会暴露在任何地方。一旦有更好的支持通过 Metrics 系统发布这些数字,我们应该启用统计。
原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/queryable_state.html