Flink 流 - 延迟和吞吐量检测答案

【问题标题】：Flink streaming - latency and throughput detectionFlink 流 - 延迟和吞吐量检测
【发布时间】：2017-10-29 21:11:10
【问题描述】：

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

（完整场景：我已通过 Producer 将消息作为 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();
}

【问题讨论】：

标签： json apache-kafka apache-flink kafka-producer-api flink-streaming

【解决方案1】：

This benchmarking application 可能是一个不错的起点。 latency tracking 和 metrics available from Flink's Kafka connector 上的文档也应该是有趣的阅读。

【讨论】：

【解决方案2】：

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

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

最后，对于延迟跟踪，也许您可以尝试这里解释的内容 - latency-tracking，同样您可以使用 - meters 获得吞吐量

【讨论】：