Kafka + Spark Streaming：1 秒的恒定延迟答案

【问题标题】：Kafka + Spark Streaming: constant delay of 1 secondKafka + Spark Streaming：1 秒的恒定延迟
【发布时间】：2017-11-04 04:22:38
【问题描述】：

EDIT2：最后我使用 Java 制作了自己的生产者，它运行良好，所以问题出在 Kafka-console-producer。 kafka-console-consumer 运行良好。

编辑：我已经尝试使用 0.9.0.1 版本并且具有相同的行为。

我正在完成我的学士期末项目，即 Spark Streaming 和 Flink 之间的比较。在这两个框架之前，我使用 Kafka 和脚本来生成数据（如下所述）。我的第一个测试是将两个框架之间的延迟与简单的工作负载进行比较，Kafka 给了我一个非常高的延迟（持续 1 秒）。为简单起见，目前我只在一台机器上运行 Kafka 和 Spark。

我已经寻找并发现了类似的问题，并尝试了他们提供的解决方案，但没有任何改变。我还检查了官方文档中的所有 Kafka 配置，并将延迟的重要信息放在我的配置文件中，这是我的配置：

Kafka 0.10.2.1 - Spark 2.1.0

server.properties：

num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
num.partitions=2
num.recovery.threads.per.data.dir=1
log.flush.interval.messages=1000
log.flush.interval.ms=50
log.retention.hours=24
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
flush.messages=100
flush.ms=10

producer.properties：

compression.type=none
max.block.ms=200
linger.ms=50
batch.size=0

Spark Streaming 程序：（打印接收到的数据，以及数据创建时间和函数处理时间之间的差异）

package com.tfg.spark1.spark1;

import java.util.Map;
import java.util.HashMap;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;
import org.apache.spark.streaming.kafka.*;

public final class Timestamp {

    public static void main(String[] args) throws Exception {
        if (args.length < 2) {
            System.err.println("Usage: Timestamp <topics> <numThreads>");
            System.exit(1);
        }

        SparkConf conf = new SparkConf().setMaster("spark://192.168.0.155:7077").setAppName("Timestamp");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.milliseconds(100));


        Map<String, Integer> topicMap = new HashMap<String, Integer>();
        int numThreads = Integer.parseInt(args[1]);
        topicMap.put(args[0], numThreads);

        JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, "192.168.0.155:2181", "grupo-spark", topicMap); //Map<"test", 2>

        JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
            private static final long serialVersionUID = 1L;

            public String call (Tuple2<String, String> tuple2) {
                return tuple2._2();
            }
        });

        JavaDStream<String> newLine = lines.map(new Function<String, String>() {
            private static final long serialVersionUID = 1L;

            public String call(String line) {
                String[] tuple = line.split(" ");
                String totalTime = String.valueOf(System.currentTimeMillis() - Long.valueOf(tuple[1]));
                //String newLine = line.concat(" " + String.valueOf(System.currentTimeMillis()) + " " + totalTime);

                return totalTime;
            }
        });

        lines.print();
        newLine.print();

        jssc.start();
        jssc.awaitTermination();
    }
}

生成的数据具有以下格式：

"Random bits" + " " + "current time in ms"
01 1496421618634
11 1496421619044
00 1496421619451
00 1496421618836
10 1496421619247

最后，当我运行我的 Spark Streaming 程序和脚本生成器（每 200 毫秒生成一次数据）时，Spark（批处理间隔=100 毫秒）会打印 9 个空批处理，并且每秒（总是 900 毫秒，如下例所示：时间： 1496421619900 毫秒）这个结果：

-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
01 1496421618634
11 1496421619044
00 1496421619451
00 1496421618836
10 1496421619247
-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
1416
1006
599
1214
803

另外，如果我运行一个 Kafka 命令行生产者和另一个命令行消费者，在消费者中打印生成的数据总是需要一些时间。

提前感谢您的帮助！

【问题讨论】：

首先尝试简单的消费者，看看它是特定于火花还是特定于卡夫卡。报告 30 毫秒延迟的帖子（也来自linkedin）很少。
你的意思是 kafka-console-consumer 吗？我已经尝试过，它也延迟接收元素。我还从几个站点了解到它可以实现这种延迟。我也会尝试使用较旧的 Kafka 版本。谢谢！ :D
这也可能取决于您的硬件（例如线程数）。还要尝试看到系统处于稳定状态（不仅仅是一两条消息）可能需要时间来“热身”
我在 Kafka 中使用 3 个线程进行网络连接，8 个线程用于磁盘 I/O，如您在我的配置中所见。无论如何，我一直在用非常小的数据尝试这个，所以它不应该是一个问题。我创建了一个 bash 脚本，它每 X 毫秒发送一行数据，我让它工作了超过一分钟，所以恐怕这不是问题 :(

标签： apache-spark apache-kafka delay spark-streaming kafka-producer-api

【解决方案1】：

我刚刚更新了您打开的 JIRA，说明您总是看到 1000 毫秒延迟的原因。

https://issues.apache.org/jira/browse/KAFKA-5426

我在这里报告原因......

linger.ms 参数是使用命令行上的--timeout 选项设置的，如果未指定，则为 1000 毫秒。同时使用命令行上的--max-partition-memory-bytes 选项设置batch.size 参数，如果未指定，则为16384。这意味着即使您使用 --producer-property 或 --producer.config 指定 linger.ms 和 batch.size，它们也将始终被上述“特定”选项覆盖。

【讨论】：