Flink从1.2.0开始支持可查询状态(Queryable State)服务。可查询状态，允许用户从外部系统(如业务系统)查询Flink作业内部的状态。

Queryable State服务架构

image.png

  1. QueryableStateClient: 客户端。运行在外部系统。提交查询请求并接收最终返回的结果。
  2. QueryableStateClientProxy: 客户端代理。运行在每个TaskManager上。接收客户端的请求，找到Key对应的TaskManager，然后将请求转发给具体的查询服务，并负责最终向客户端返回结果。
  3. QueryableStateServer: 查询服务。运行在每个TaskManager上。处理来自客户端代理的请求并返回结果。

激活Queryable State服务

Flink On Yarn模式

Flink On Yarn 集群模式可通过如下设置启动。

  1. 添加依赖

    1. cp ${FLINK_HOME}/opt/flink-queryable-state-runtime_2.11-1.13.0.jar ${FLINK_HOME}/lib/

  2. 启用Queryable State服务
    在${FLINK_HOME}/conf/flink-conf.yaml中设置queryable-state.enable: true

  3. 查看TaskManager日志: 在TaskManager中见如下日志，即启动。

    1. Started Queryable State Server @ /x.x.x.x:9067.
    2. Started Queryable State Proxy Server @ /x.x.x.x:9069

    Flink Local模式

    Flink Local模式可通过如下设置启动。

  4. 添加依赖

    1. <dependency>
    2.  <groupId>org.apache.flink</groupId>
    3.  <artifactId>flink-queryable-state-runtime_2.11</artifactId>
    4.  <version>1.13.0</version>
    5. </dependency>

  5. 启用Queryable State服务 ```java Configuration config = new Configuration(); config.setInteger(ConfigOptions.key(“rest.port”).defaultValue(8081),8081); config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true); //启用Queryable State服务 config.setBoolean(QueryableStateOptions.ENABLE_QUERYABLE_STATE_PROXY_SERVER, true); StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);

  2. 3. 查看TaskManager日志: 在TaskManager中见如下日志，即启动。
  3. ```java
  4.  Started Queryable State Server @ /127.0.0.1:9067
  5.  Started Queryable State Proxy Server @ /127.0.0.1:9069

使状态可查询

通过以上设置，已在Flink集群上激活了可查询状态服务，除此之外，还需要在代码中暴露具体的可查询状态。有两种方式:

  1. 将DataStream转换为QueryableStateStream。如将KeyedStream转换为QueryableStateStream，即可设定KeyedStream中所有Key的State可查。
  2. 通过状态描述StateDescriptor的setQueryable(String queryableStateName)方法，可设定某个Key的State可查。

以下通过两个例子总结这两种方式的使用。

主要依赖

Server端

  1.     <dependency>
  2.       <groupId>org.apache.flink</groupId>
  3.       <artifactId>flink-streaming-java_2.11</artifactId>
  4.       <version>1.13.0</version>
  5.     </dependency>
  7.     <dependency>
  8.       <groupId>org.apache.flink</groupId>
  9.       <artifactId>flink-runtime-web_2.11</artifactId>
  10.       <version>1.13.0</version>
  11.     </dependency>
  13.     <dependency>
  14.       <groupId>org.apache.flink</groupId>
  15.       <artifactId>flink-queryable-state-runtime_2.11</artifactId>
  16.       <version>1.13.0</version>
  17.     </dependency>

Client端

  1.     <dependency>
  2.       <groupId>org.apache.flink</groupId>
  3.       <artifactId>flink-core</artifactId>
  4.       <version>1.9.0</version>
  5.     </dependency>
  7.     <dependency>
  8.       <groupId>org.apache.flink</groupId>
  9.       <artifactId>flink-queryable-state-client-java</artifactId>
  10.       <version>1.9.0</version>
  11.     </dependency>

通过Queryable State Stream使状态可查

Server端(Flink Job)

  1. package com.lingyao.state;
  3. import org.apache.flink.api.common.functions.FlatMapFunction;
  4. import org.apache.flink.api.java.functions.KeySelector;
  5. import org.apache.flink.api.java.tuple.Tuple2;
  6. import org.apache.flink.api.java.utils.ParameterTool;
  7. import org.apache.flink.configuration.ConfigConstants;
  8. import org.apache.flink.configuration.Configuration;
  9. import org.apache.flink.configuration.QueryableStateOptions;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  13. import org.apache.flink.streaming.api.windowing.time.Time;
  14. import org.apache.flink.util.Collector;
  16. /**
  17.  * @Author : wangxiaohui
  18.  * @Date: 2021-09-29 11:26
  19.  * Queryable State Stream
  20.  */
  21. public class QueryableStateByQueryableStateStream {
  22.     public static void main(String[] args) throws Exception {
  24.         // 设置参数
  25.         ParameterTool parameterTool = ParameterTool.fromArgs(args);
  26.         final String host = parameterTool.get("host", "localhost");
  27.         final int port = parameterTool.getInt("port", 1234);
  28.         final int parallelism = parameterTool.getInt("parallelism", 4);
  30.         // 配置环境
  31.         Configuration config = new Configuration();
  32.         config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
  33.         config.setBoolean(QueryableStateOptions.ENABLE_QUERYABLE_STATE_PROXY_SERVER, true);
  35.         StreamExecutionEnvironment env =
  36.                 StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
  37.         env.getConfig().setAutoWatermarkInterval(0);
  38.         env.setParallelism(parallelism);
  40.         // 输入数据源：每行数据格式 event,pv
  41.         SingleOutputStreamOperator<Tuple2<String, Long>> source =
  42.                 env.socketTextStream(host, port)
  43.                         .flatMap(
  44.                                 new FlatMapFunction<String, Tuple2<String, Long>>() {
  45.                                     @Override
  46.                                     public void flatMap(String value, Collector<Tuple2<String, Long>> out) {
  48.                                         String[] splits = value.trim().split(",");
  50.                                         out.collect(new Tuple2<>(splits[0], Long.valueOf(splits[1])));
  51.                                     }
  52.                                 });
  54.         // 窗口统计: 最近5秒钟内，每个事件的最大pv
  55.         SingleOutputStreamOperator<Tuple2<String, Long>> result =
  56.                 source
  57.                         .keyBy(
  58.                                 new KeySelector<Tuple2<String, Long>, String>() {
  59.                                     @Override
  60.                                     public String getKey(Tuple2<String, Long> value) throws Exception {
  61.                                         return value.f0;
  62.                                     }
  63.                                 })
  64.                         .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
  65.                         .max(1);
  67.         // 输出结果
  68.         result.print();
  70.         // 使得结果的状态可查
  71.         // asQueryableState 返回QueryableStateStream
  72.         // QueryableStateStream类似于一个接收器，无法进行进一步转换
  73.         // QueryableStateStream接收传入的数据并更新状态
  74.         result
  75.                 .keyBy(
  76.                         new KeySelector<Tuple2<String, Long>, String>() {
  77.                             @Override
  78.                             public String getKey(Tuple2<String, Long> value) throws Exception {
  79.                                 return value.f0;
  80.                             }
  81.                         })
  82.                 .asQueryableState("lastFiveSecondsMaxPV");
  84.         env.execute();
  85.     }
  86. }

Client端

  1. package com.lingyao.state;
  3. import org.apache.flink.api.common.JobID;
  4. import org.apache.flink.api.common.state.ValueState;
  5. import org.apache.flink.api.common.state.ValueStateDescriptor;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeHint;
  8. import org.apache.flink.api.common.typeinfo.TypeInformation;
  9. import org.apache.flink.api.java.tuple.Tuple2;
  10. import org.apache.flink.queryablestate.client.QueryableStateClient;
  12. import java.util.concurrent.CompletableFuture;
  14. /**
  15.  * @Author : wangxiaohui
  16.  * @Date: 2021-09-29 11:29
  17.  */
  18. public class QueryClient {
  19.     public static void main(String[] args) throws Exception {
  21.         if (args.length == 0) {
  22.             throw new IllegalArgumentException(
  23.                     "Missing Flink Job ID ...\n"
  24.                             + "Usage: ./QueryClient <jobID> [jobManagerHost] [jobManagerPort]");
  25.         }
  27.         final JobID jobId = JobID.fromHexString(args[0]);
  28.         final String jobManagerHost = args.length > 1 ? args[1] : "localhost";
  29.         final int jobManagerPort = args.length > 1 ? Integer.parseInt(args[1]) : 9069;
  31.         QueryableStateClient client = new QueryableStateClient(jobManagerHost, jobManagerPort);
  33.     /*
  34.       状态描述符
  35.       event_five_seconds_max_pv 状态名称，Server端此名称是通过UUID生成的，这里随意即可
  36.       TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {})) 状态中值的类型，应和Server端相同
  37.      */
  38.         ValueStateDescriptor<Tuple2<String, Long>> stateDescriptor =
  39.                 new ValueStateDescriptor<>(
  40.                         "lastFiveSecondsMaxPV",
  41.                         TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {}));
  42.         final String key = "click";
  44.     /*
  45.      轮询执行查询
  46.      event_five_seconds_max_pv queryableStateName，应和Server端的queryableStateName相同
  47.     */
  48.         while (true) {
  49.             CompletableFuture<ValueState<Tuple2<String, Long>>> completableFuture =
  50.                     client.getKvState(
  51.                             jobId,
  52.                             "lastFiveSecondsMaxPV",
  53.                             key,
  54.                             BasicTypeInfo.STRING_TYPE_INFO,
  55.                             stateDescriptor);
  57.             System.out.println(completableFuture.get().value());
  59.             Thread.sleep(1000);
  60.         }
  61.     }
  62. }

通过StateDescriptor的setQueryable方法使状态可查

Server端(Flink Job)

  1. package com.lingyao.state;
  4. import org.apache.flink.api.common.functions.FlatMapFunction;
  5. import org.apache.flink.api.common.state.ValueState;
  6. import org.apache.flink.api.common.state.ValueStateDescriptor;
  7. import org.apache.flink.api.common.typeinfo.Types;
  8. import org.apache.flink.api.java.functions.KeySelector;
  9. import org.apache.flink.api.java.tuple.Tuple2;
  10. import org.apache.flink.api.java.tuple.Tuple5;
  11. import org.apache.flink.api.java.utils.ParameterTool;
  12. import org.apache.flink.configuration.ConfigConstants;
  13. import org.apache.flink.configuration.Configuration;
  14. import org.apache.flink.configuration.QueryableStateOptions;
  15. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  16. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  17. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  18. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  19. import org.apache.flink.streaming.api.windowing.time.Time;
  20. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  21. import org.apache.flink.util.Collector;
  22. import org.joda.time.DateTime;
  23. import org.joda.time.DateTimeZone;
  25. /**
  26.  * @Author : wangxiaohui
  27.  * @Date: 2021-09-29 14:36
  28.  */
  29. public class QueryableStateByStateDescriptor {
  31.     public static void main(String[] args) throws Exception {
  33.         // 设置参数
  34.         ParameterTool parameterTool = ParameterTool.fromArgs(args);
  35.         final String host = parameterTool.get("host", "localhost");
  36.         final int port = parameterTool.getInt("port", 1234);
  37.         final int parallelism = parameterTool.getInt("parallelism", 4);
  39.         // 配置环境
  40.         Configuration config = new Configuration();
  41.         config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
  42.         config.setBoolean(QueryableStateOptions.ENABLE_QUERYABLE_STATE_PROXY_SERVER, true);
  44.         StreamExecutionEnvironment env =
  45.                 StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
  46.         env.setParallelism(parallelism);
  47.         env.getConfig().setAutoWatermarkInterval(0);
  49.         // 输入数据源：当前action产生时就上报一条数据，数据格式为 action,money
  50.         SingleOutputStreamOperator<Tuple2<String, Long>> source =
  51.                 env.socketTextStream(host, port)
  52.                         .flatMap(
  53.                                 new FlatMapFunction<String, Tuple2<String, Long>>() {
  54.                                     @Override
  55.                                     public void flatMap(String value, Collector<Tuple2<String, Long>> out) {
  57.                                         String[] splits = value.trim().split(",");
  59.                                         out.collect(new Tuple2<>(splits[0], Long.valueOf(splits[1])));
  60.                                     }
  61.                                 });
  63.         // 窗口统计: 计算最近5秒钟内，每个action的count(money)数、sum(money)数
  64.         SingleOutputStreamOperator<Tuple5<String, String, String, Long, Long>> result =
  65.                 source
  66.                         .keyBy(
  67.                                 new KeySelector<Tuple2<String, Long>, String>() {
  68.                                     @Override
  69.                                     public String getKey(Tuple2<String, Long> value) throws Exception {
  70.                                         return value.f0;
  71.                                     }
  72.                                 })
  73.                         .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
  74.                         .process(new CustomCountSumWindowFunction());
  76.         result.print();
  78.         env.execute();
  79.     }
  81.     /** 自定义WindowFunction,统一每个Key对应的窗口的count数、sum数 */
  82.     static class CustomCountSumWindowFunction
  83.             extends ProcessWindowFunction<
  84.                         Tuple2<String, Long>, Tuple5<String, String, String, Long, Long>, String, TimeWindow> {
  86.         // 定义一个ValueState，来存放状态
  87.         private transient ValueState<Tuple5<String, String, String, Long, Long>> sumCountValueState;
  89.         /**
  90.          * 算子CustomCountSumWindowFunction实例化时，只执行一次
  91.          *
  92.          * @param parameters
  93.          * @throws Exception
  94.          */
  95.         @Override
  96.         public void open(Configuration parameters) throws Exception {
  97.             ValueStateDescriptor<Tuple5<String, String, String, Long, Long>> valueStateDescriptor =
  98.                     new ValueStateDescriptor<>(
  99.                             "lastFiveSecondsCountSumValueState",
  100.                             Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG, Types.LONG));
  102.             // 通过ValueStateDescriptor.setQueryable 开放此状态,使此状态可查
  103.             valueStateDescriptor.setQueryable("lastFiveSecondsCountSum");
  104.             sumCountValueState = getRuntimeContext().getState(valueStateDescriptor);
  105.         }
  107.         /**
  108.          * 每个窗口都会调用一次process方法
  109.          *
  110.          * @param key 当前窗口对应的Key
  111.          * @param context 窗口上下文
  112.          * @param elements 当前窗口的所有元素
  113.          * @param out 收集输出记录
  114.          * @throws Exception
  115.          */
  116.         @Override
  117.         public void process(
  118.                 String key,
  119.                 Context context,
  120.                 Iterable<Tuple2<String, Long>> elements,
  121.                 Collector<Tuple5<String, String, String, Long, Long>> out)
  122.                 throws Exception {
  124.             // 计算count数、sum数
  125.             long currentCount = 0L;
  126.             long currentSum = 0L;
  128.             for (Tuple2<String, Long> value : elements) {
  129.                 currentCount += 1;
  130.                 currentSum += value.f1;
  131.             }
  133.             // 获取Window开始时间、结束时间
  134.             TimeWindow window = context.window();
  135.             String windowStartTime =
  136.                     new DateTime(window.getStart(), DateTimeZone.UTC).toString("yyyy-MM-dd HH:mm:ss");
  137.             String windowEndTime =
  138.                     new DateTime(window.getEnd(), DateTimeZone.UTC).toString("yyyy-MM-dd HH:mm:ss");
  140.             // 得到当前值
  141.             Tuple5<String, String, String, Long, Long> currentValue =
  142.                     new Tuple5<>(key, windowStartTime, windowEndTime, currentCount, currentSum);
  144.             // 更新状态
  145.             sumCountValueState.update(currentValue);
  147.             // 输出结果
  148.             out.collect(currentValue);
  149.         }
  150.     }
  151. }

Client端

  1. package com.lingyao.state;
  3. import org.apache.flink.api.common.JobID;
  4. import org.apache.flink.api.common.state.ValueState;
  5. import org.apache.flink.api.common.state.ValueStateDescriptor;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.Types;
  8. import org.apache.flink.api.java.tuple.Tuple5;
  9. import org.apache.flink.queryablestate.client.QueryableStateClient;
  11. import java.util.concurrent.CompletableFuture;
  13. /**
  14.  * @Author : wangxiaohui
  15.  * @Date: 2021-09-29 14:42
  16.  */
  17. public class QueryClient2 {
  18.     public static void main(String[] args) throws Exception {
  20.         if (args.length == 0) {
  21.             throw new IllegalArgumentException(
  22.                     "Missing Flink Job ID ...\n"
  23.                             + "Usage: ./QueryClient <jobID> [jobManagerHost] [jobManagerPort]");
  24.         }
  26.         final JobID jobId = JobID.fromHexString(args[0]);
  27.         final String jobManagerHost = args.length > 1 ? args[1] : "localhost";
  28.         final int jobManagerPort = args.length > 1 ? Integer.parseInt(args[1]) : 9069;
  30.         QueryableStateClient client = new QueryableStateClient(jobManagerHost, jobManagerPort);
  32.     /*
  33.      状态描述符
  34.     */
  35.         ValueStateDescriptor<Tuple5<String, String, String, Long, Long>> valueStateDescriptor =
  36.                 new ValueStateDescriptor<>(
  37.                         "lastFiveSecondsCountSumValueState",
  38.                         Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG, Types.LONG));
  40.         final String key = "click";
  42.     /*
  43.      轮询执行查询
  44.      lastFiveSecondsCountSum queryableStateName，应和Server端的queryableStateName相同
  45.     */
  46.         while (true) {
  48.             CompletableFuture<ValueState<Tuple5<String, String, String, Long, Long>>> completableFuture =
  49.                     client.getKvState(
  50.                             jobId,
  51.                             "lastFiveSecondsCountSum",
  52.                             key,
  53.                             BasicTypeInfo.STRING_TYPE_INFO,
  54.                             valueStateDescriptor);
  56.             Tuple5<String, String, String, Long, Long> value = completableFuture.get().value();
  57.             System.out.println(
  58.                     "Key: "
  59.                             + value.f0
  60.                             + " ,WindowStart: "
  61.                             + value.f1
  62.                             + " ,WindowEnd: "
  63.                             + value.f2
  64.                             + " ,Count: "
  65.                             + value.f3
  66.                             + " ,Sum: "
  67.                             + value.f4);
  69.             Thread.sleep(1000);
  70.         }
  71.     }
  72. }

TODO

将状态保存在外部系统。
修改Checkpoint中的状态，让应用根据修改后的状态启动。
原文链接：https://blog.csdn.net/wangpei1949/article/details/100608828