3.5、Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 Queryable State Beta（可查询状态）

本是古典何须时尚 2024-04-17 17:18 14阅读 0赞

**目录**

状态结构

Activating Queryable State（激活可查询状态）

Making State Queryable（配置状态查询）

Queryable State Stream

Managed Keyed State

Querying State

Example

Configuration

State Server

Proxy

局限性

--------------------

> Note：用于 queryable state 的客户端API目前处于不断发展的状态，并且不能保证所提供接口的稳定性。在即将到来的Flink版本中，很可能会重构。

简而言之，该特性向外部公开了Flink的 managed keyed (partitioned) state (请参阅：[Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 State的工作原理][Flink_Stream API_- State _ Fault Tolerance_ State])，并允许用户从 Flink 外部查询一个 job 的 state。 对于某些场景，queryable state 消除了与外部系统（如 key-value 存储）进行分布式操作/事务的需要，而这些系统通常是生产中的瓶颈。此外对于调试程序也是很有用的。

> Attention：在查询状态对象时，从并发线程访问该对象而不进行任何同步或复制操作。这种设计方式是因为上面的任何一种操作都会导致作业延迟的增加。因为任何 state backend 都使用 Java heap space。例如 `MemoryStateBackend 或 FsStateBackend，`在检索值时不复制，而是直接引用存储值，read-modify-write 模式是不安全的，可能会导致 queryable state 服务器由于并发修改而失败。`RocksDBStateBackend 不会有上述问题。`

# 状态结构 #

下面展示如何使用 Queryable State，简要描述他的实现类：

*  `QueryableStateClient：在集群外提交用户的查询请求`
 *  `QueryableStateClientProxy：运行在集群中每个 TM 中，负责接收客户端的查询请求，代替 TM 获取请求状态，然后再发给客户端`
 *  `QueryableStateServer：运行在每个 TM 上，`负责为本地存储的状态提供服务。

客户端连接到其中一个代理，并发送一个与特定密钥相关联的状态请求，keyed state 在 Key Groups中，每个 TM 中都有这些 key groups，通过访问 JM 获取哪个 TM 中有这个 k 的 keygroups。根据结果，代理将查询运行在 TM 上的 QueryableStateServer，以获得与 k 关联的状态，并将响应结果转发回客户端。

# Activating Queryable State（激活可查询状态） #

要在Flink集群上启用 queryable state，需要执行以下操作：

1.  将 `flink-queryable-state-runtime_2.11-1.9.0.jar `从Flink发行版的 opt/ 文件夹复制到lib/文件夹。
2.  `queryable-state.enable 设置为 ture，详细的请查阅：`[Flink部署运维（Deployment & Operations）- Configuration（Flink配置项）][Flink_Deployment _ Operations_- Configuration_Flink]

验证集群是否启用了该配置，可以随意找一个 TM 的日志 `"Started the Queryable State Proxy Server @ ..."`

# Making State Queryable（配置状态查询） #

现在你已经激活了集群上的 queryable state，现在看看如何使用它。为了使一个状态对外可见，需要显式地使用以下方法使其可查询：

*  `QueryableStateStream 对象可以作为 `queryable state 的入参
 *  `stateDescriptor.setQueryable(String queryableStateName) 方法将 querystate 传给 stateDescriptor`

下面给提供这两种方式的详细使用方式：

## Queryable State Stream ##

在 `KeyedStream `上调用 `.asQueryableState(stateName, stateDescriptor) `将返回一个 `QueryableStateStream`，供 queryable state 使用。

// 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)

> **Note：**没有可供查询的 `ListState` Sink ，它会导致一个不断增长的列表，这个列表可能不会被清理，最终会消耗太多内存。

返回的 QueryableStateStream 可以看作是一个 Sink，不能进一步转换。在内部，QueryableStateStream 被转换为一个算子，该算子使用所有传入的记录来更新 queryable state 实例。更新逻辑由 asQueryableState 调用中提供的 `StateDescriptor `类型实现。在下面的程序中，keyed stream 的所有记录都将通过 keyed stream 更新状态实例：

stream.keyBy(0).asQueryableState("query-name")

这类似于Scala API的 flatMapWithState。

## Managed Keyed State ##

`StateDescriptor.setQueryable(String queryableStateName)可以通过状态描述符查询`，如下面的例子所示：

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 参数可以任意选择，只用于查询。它不必与 state 的名字相同。

对于哪种状态类型可以查询，此变量没有限制。这意味着它可以用于任何 ValueState、ReduceState、ListState、MapState、AggregatingState 和当前不推荐使用的 FoldingState。

## Querying State ##

到目前为止，您已经将集群设置为使用可查询状态运行，并且已经将(一些)状态声明为可查询状态。现在是查看如何查询此状态的时候了。

为此，您可以使用 QueryableStateClient 帮助类。这在 flink-queryable-state-client jar 中可用，它必须与 flink-core 一起作为依赖项显式地包含在项目的 pom.xml 中，如下所示:

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-core</artifactId>
      <version>1.9.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-queryable-state-client-java</artifactId>
      <version>1.9.0</version>
    </dependency>

有关更多信息，请查阅：[Flink基础之配置Maven依赖、连接、库][Flink_Maven]。

QueryableStateClient 将查询提交给内部代理，然后该代理将处理查询并返回最终结果。初始化客户机的唯一要求是提供一个有效的 TaskManager 主机名(请记住，每个 TM 上都运行着一个可查询的状态代理)和代理侦听的端口。有关如何配置代理和状态服务器端口的更多信息，请参见下方config。

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`，最终包含可查询状态实例的状态值，该实例由 `queryableStateName 的 jobID `标识。keyTypeInfo 将告诉 Flink 如何序列化/反序列化它。最后，`stateDescriptor `包含关于请求状态的必要信息，即它的类型(`Value`, `Reduce `等)和关于如何序列化/反序列化它的必要信息。

细心的读者会注意到返回的 future 包含一个类型 S 的值，即一个包含实际值的状态对象。这可以是 Flink 支持的任何状态类型：ValueState、ReduceState、ListState、MapState、AggregatingState 和当前不推荐使用的 FoldingState。

> 注意：这些状态对象不允许修改所包含的状态。您可以使用它们来获取状态的实际值，例如使用 valueState.get()，或者迭代包含的 <K, V> 的记录，例如使用 mapState.entries()，但是您不能修改它们。例如，在返回的列表状态上调用 add() 方法将引发 UnsupportedOperationException。

> 注意：客户机是异步的，可以由多个线程共享。在未使用时，需要通过 QueryableStateClient.shutdown()关闭它，以便释放资源。

## Example ##

下面的示例扩展了 CountWindowAverage 示例(详细请查阅：[Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 State的工作原理][Flink_Stream API_- State _ Fault Tolerance_ State 1])，使其可查询，并展示了如何查询这个值：

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);
        }
    }

一旦在作业中使用，您可以检索作业ID，然后从该操作符查询任何键的当前状态：

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();
            }
    });

## Configuration ##

以下配置参数影响 queryable state server 和 client 的行为。它们在 QueryableStateOptions 中定义。

### State Server ###

*  `queryable-state.server.ports：`queryable state server 的服务器端口范围。如果同一台机器上运行多个 TM，这对避免端口冲突非常有用。指定的范围可以是：一个端口:“9123”，一个端口范围:“50100-50200”，或者一个范围和点列表:“50100-50200,50300-50400,51234”。默认端口是9067。 
 *  `queryable-state.server.network-threads：`接收状态服务器传入请求的网络(事件循环)线程数(0 => \#slots)，范围为0-solts的数量之间，下同！！！
 *  `queryable-state.server.query-threads：`处理/服务状态服务器传入请求的线程数(0 => \#slots)。

### Proxy ###

*  `queryable-state.proxy.ports：`queryable state server 代理的服务器端口范围。如果同一台机器上运行多个 TM，这对避免端口冲突非常有用。指定的范围可以是:一个端口:“9123”，一个端口范围:“50100-50200”，或者一个范围和点列表:“50100-50200,50300-50400,51234”。默认端口是9069。
 *  `queryable-state.proxy.network-threads：`接收客户端代理传入请求的网络(事件循环)线程数(0 => \#slots)
 *  `queryable-state.proxy.query-threads：`为客户机代理处理/服务传入请求的线程数(0 => \#slots)。

### 局限性 ###

*  可查询状态生命周期绑定到作业的生命周期，例如，任务在启动时注册可查询状态，在处理时注销可查询状态。在未来的版本中，最好将其解耦，以便允许在任务完成后进行查询，并通过状态复制加速恢复。
 *  关于可用 KvState 的通知通过一个简单的 tell 。在未来，这一点应该得到改进，使其在提出要求和表示感谢方面更加有力。
 *  服务器和客户机跟踪查询的统计信息。由于它们不会在任何地方公开，因此默认情况下它们是禁用的。一旦有更好的支持通过 Metrics 系统发布这些数字，我们就应该启用统计数据。

[Flink_Stream API_- State _ Fault Tolerance_ State]: https://blog.csdn.net/zpf_940810653842/article/details/100030816
[Flink_Deployment _ Operations_- Configuration_Flink]: https://blog.csdn.net/zpf_940810653842/article/details/99873275
[Flink_Maven]: https://blog.csdn.net/zpf_940810653842/article/details/99622787
[Flink_Stream API_- State _ Fault Tolerance_ State 1]: https://blog.csdn.net/zpf_940810653842/article/details/100030816#using-managed-keyed-state