edit2:最后我用java制作了自己的producer,效果很好,所以问题出在kafka控制台producer上。Kafka控制台消费者工作得很好。
编辑:我已经尝试过版本0.9.0.1,并且具有相同的行为。
我正在做我的学士学位期末专题,比较spark streaming和flink。在这两个框架之前,我使用kafka和一个脚本来生成数据(解释如下)。我的第一个测试是比较两个具有简单工作负载的框架之间的延迟,kafka给了我一个非常高的延迟(持续1秒)。为了简单起见,目前我只在一台机器上运行Kafka和Spark。
我已经寻找并发现了类似的问题,并尝试了他们给出的解决方案,但没有任何改变。我还检查了官方文档中的所有Kafka配置,并将延迟的重要信息放在我的配置文件中,这是我的配置:
Kafka0.10.2.1-Spark2.1.0
服务器属性:
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
生产商属性:
compression.type=none
max.block.ms=200
linger.ms=50
batch.size=0
spark streaming program:(打印接收到的数据,以及创建数据和处理函数的时间之间的差异)
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流程序和脚本生成器时,它每200ms生成一次数据,spark(批间隔=100ms)打印9个空批,并且每秒钟(总是900ms的时刻,如本例:time:1496421619900ms)结果如下:
-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
01 1496421618634
11 1496421619044
00 1496421619451
00 1496421618836
10 1496421619247
-------------------------------------------
Time: 1496421619900 ms
-------------------------------------------
1416
1006
599
1214
803
另外,如果我运行一个kafka命令行生产者和另一个命令行使用者,那么在使用者中打印生成的数据总是需要一些时间。
提前感谢您的帮助!
1条答案
按热度按时间jv2fixgn1#
我刚刚更新了你打开的jira,解释了为什么你总是看到1000毫秒的延迟。
https://issues.apache.org/jira/browse/kafka-5426
我在这里报告原因。。。
linger.ms参数是使用
--timeout
命令行上的选项,如果未指定,则为1000 ms。同时,使用--max-partition-memory-bytes
命令行上的选项,如果未指定,则为16384。这意味着,即使使用--producer property或--producer.config指定linger.ms和batch.size,它们也将始终被上述“特定”选项覆盖。