Kafka + Spark Streaming:持续1秒的延迟 [英] Kafka + Spark Streaming: constant delay of 1 second
问题描述
EDIT2 :最后，我使用Java制作了自己的制作器，并且效果很好，所以问题出在Kafka-console-producer 中. kafka-console-consumer效果很好.
Finally I have made my own producer using Java and it works well, so the problem is in the Kafka-console-producer. The kafka-console-consumer works well.
编辑:我已经尝试过0.9.0.1版，并且具有相同的行为.
I have already tried with the version 0.9.0.1 and has the same behaviour.
我正在完成学士学位的最终项目，即Spark Streaming和Flink之间的比较.在这两个框架之前，我都使用Kafka和一个脚本来生成数据(如下所述).我的第一个测试是比较两个框架与简单工作负载之间的延迟，而Kafka给我的延迟确实很高(持续1秒).为简单起见，目前，我仅在Kafka和Spark上运行一台计算机.
I am working on my bachelor's final project, a comparison between Spark Streaming and Flink. Before both frameworks I am using Kafka and a script to generate the data (explained below). My first test is to compare the latency between both frameworks with simple workloads and Kafka is giving me a really high latency (1 second constantly). For simplicity, for the moment I am running in only one machine both Kafka and Spark.
我已经寻找并发现了类似的问题，并尝试了他们给出的解决方案，但没有任何改变.我还检查了官方文档中的所有Kafka配置，并将延迟的重要信息放在我的配置文件中，这是我的配置:
I have already looked for and found similar problems, and tried the solutions they give but nothing changed. I have checked also all the Kafka configurations in the official documentation and put the importants for the latency in my config files, this is my configuration:
Kafka 0.10.2.1-Spark 2.1.0
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程序:(打印接收到的数据，以及该数据的创建时间和处理时间之间的差异)
Spark Streaming program: (which prints the received data, and the difference between when the data was created and when is being processed for the function)
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程序和脚本生成器时，该脚本生成器每200ms生成一次数据，Spark(批处理间隔= 100ms)打印9个空批处理，并每秒(总是900ms的时间)打印此示例，例如: 1496421619 900 毫秒)此结果:
Finally when I run my Spark Streaming program and the script generator, which generates the data every 200ms, Spark (batch interval=100ms) prints 9 empty batches, and every second (always 900ms moment, like in this example: Time: 1496421619900 ms) this results:
-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
01 1496421618634
11 1496421619044
00 1496421619451
00 1496421618836
10 1496421619247
-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
1416
1006
599
1214
803
另外，如果我运行一个Kafka命令行生产者和另一个命令行消费者，则总是花费一些时间在消费者中打印生产的数据.
Also if I run one Kafka command-line-producer and another command-line-consumer, it always takes some time to print the produced data in the consumer.
提前感谢您的帮助！
推荐答案
我刚刚更新了您打开的JIRA，原因是您总是看到1000毫秒的延迟.
I have just updated the JIRA you opened with the reason why you always see the 1000 ms delay.
https://issues.apache.org/jira/browse/KAFKA-5426
我在这里报告原因...
I report here the reason ...
在命令行上使用--timeout选项设置linger.ms参数，如果未指定，则为1000 ms.
同时，在命令行上使用--max-partition-memory-bytes选项设置batch.size参数，如果未指定，则为16384.
这意味着即使您使用--producer-property或--producer.config指定linger.ms和batch.size，它们也始终会被上述特定"选项覆盖.
the linger.ms parameter is set using the --timeout option on the command line which if not specified is 1000 ms.
At same time the batch.size parameter is set using the --max-partition-memory-bytes option on the command line which if not specified is 16384.
It means that even if you specify linger.ms and batch.size using --producer-property or --producer.config, they will be always overwritten by the above "specific" options.
这篇关于Kafka + Spark Streaming:持续1秒的延迟的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
