java - 如何计时 Apache Flink 流中的检查点?
问题描述
我正在运行以 RocksDB 作为状态后端的 Apache Flink 的欺诈检测器示例。我想知道 Apache Flink 检查点状态需要多长时间。
我的方法是在检查点功能之前和之后打印时间。
我找不到检查点的函数/类或任何代码段,我尝试通过 IDE 调试但徒劳无功。
这是我到目前为止所经历的:
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package spendreport;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.source.TransactionSource;
//org.apache.flink.contrib.streaming.state
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import javax.security.auth.login.Configuration;
/**
* Skeleton code for the datastream walkthrough
*/
public class FraudDetectionJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// env.setStateBackend(new RocksDBStateBackend(filebackend, true));
// Enabling Checkpoint
long checkpointInterval = 5000;
env.enableCheckpointing(checkpointInterval);
// Enable Web UI
// Configuration conf = new Configuration();
// env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
DataStream<Transaction> transactions = env
.addSource(new TransactionSource())
.name("transactions");
DataStream<Alert> alerts = transactions
.keyBy(Transaction::getAccountId)
.process(new FraudDetector())
.name("fraud-detector");
alerts
.addSink(new AlertSink())
.name("send-alerts");
env.execute("Fraud Detection");
}
}
我进入该execute
函数并找到代码为检查点设置配置的许多地方(例如检查超时间隔等)。但是,我找不到实际检查执行检查点的函数。
解决方案
自己衡量这一点并不容易,因为检查点部分在运行用户函数的线程中完成,部分在另一个线程中异步完成。
获取有关检查点影响的一些信息的最佳方法是查看检查点指标,这些指标被方便地收集在一起并显示在 Web UI 中:
通常,当您在 IDE 中运行时,Web UI 不可用,但您可以更改此设置:
Configuration conf = new Configuration();
env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
为此,您还必须添加此依赖项:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
推荐阅读
- javascript - 打开层 6.1。- 具有多点功能的弹出窗口
- microsoft-graph-api - MSGraph 日历事件 多个通知
- arrays - 如何通过两个变量集循环 API 调用
- javascript - 当满足特定条件时,如何使 Django 表单中的字段只读?
- android - 使用不同或相同密钥的 Android 应用程序签名?
- amazon-web-services - 如何使用 terraform 在 AWS 上启用 CORS
- docker - 将现有 heroku 应用程序的堆栈从 heroku-18 设置为 Docker 映像的“容器”?
- java - Spring Security 5.2 / WebClient 使用用户名和密码连接到另一个服务的方式是什么?
- r - 在 R - 对于所有 TRUE,所有后续 cols 都是 TRUE 吗?
- go - 在内存中编辑 zip 文件并通过 http 响应返回它会导致文件损坏