Flink流-延迟和吞吐量检测

Question

我正在尝试运行Flink流媒体作业。我想确定流处理的延迟和吞吐量。我已启动Kafka代理服务器，并收到来自Kafka的传入消息。如何计算每秒的邮件数（吞吐量）？（如rdd.count。是否有类似的方法来获取传入消息的计数）

（完整的场景：我已经通过生产者发送了消息作为Json对象。我在Json对象中添加了一些信息，如名称为字符串和System.currentTimeMills。在流式传输期间，我如何通过MessageStream（DataStream）获得发送的json对象？（

提前感谢。

代码：

/***从Kafka读取字符串并将其打印到标准输出。*/

public static void main(String[] args) throws Exception {
    System.setProperty("hadoop.home.dir", "c:/winutils/");
    // parse input argum    ents
    final ParameterTool parameterTool = ParameterTool.fromArgs(args);

    if(parameterTool.getNumberOfParameters() < 4) {
        System.out.println("Missing parameters!\nUsage: Kafka --topic <topic> " +
                "--bootstrap.servers <kafka brokers> --zookeeper.connect <zk quorum> --group.id <some id>");
        return;
    }

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().disableSysoutLogging();
    env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
    env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
    env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface

    DataStream<String> messageStream = env
            .addSource(new FlinkKafkaConsumer010<>(
                    parameterTool.getRequired("topic"),
                    new SimpleStringSchema(),
                    parameterTool.getProperties()));


    messageStream.print();

    env.execute();
}

Answer 1

这个基准测试应用程序可能是一个很好的起点。Flink的Kafka连接器提供的关于延迟跟踪和度量的文档也应该是有趣的阅读。

Answer 2

Flink UI中有一些可用的指标，您可以在其中计算每秒事件的数量等等。

您还可以添加自己的指标，根据您的需求计算一些数字，这可以显示在Flink用户界面中。

最后，对于具体的延迟跟踪，也许你可以尝试这里解释的延迟跟踪，同样，你可以使用米来获得吞吐量