我得到了火花可以流式傳輸並從只有一個分區的卡夫卡2分區主題獲取消息的情況。Spark結構化流只從卡夫卡的一個分區獲取消息
我的主題: C:\bigdata\kafka_2.11-0.10.1.1\bin\windows>kafka-topics --create --zookeeper localhost:2181 --partitions 2 --replication-factor 1 --topic test4
卡夫卡監製:
public class KafkaFileProducer {
// kafka producer
Producer<String, String> producer;
public KafkaFileProducer() {
// configs
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
//props.put("group.id", "testgroup");
props.put("batch.size", "16384");
props.put("auto.commit.interval.ms", "1000");
props.put("linger.ms", "0");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("block.on.buffer.full", "true");
// instantiate a producer
producer = new KafkaProducer<String, String>(props);
}
/**
* @param filePath
*/
public void sendFile(String filePath) {
FileInputStream fis;
BufferedReader br = null;
try {
fis = new FileInputStream(filePath);
//Construct BufferedReader from InputStreamReader
br = new BufferedReader(new InputStreamReader(fis));
int count = 0;
String line = null;
while ((line = br.readLine()) != null) {
count ++;
// dont send the header
if (count > 1) {
producer.send(new ProducerRecord<String, String>("test4", count + "", line));
Thread.sleep(10);
}
}
System.out.println("Sent " + count + " lines of data");
} catch (Exception e) {
e.printStackTrace();
}finally{
try {
br.close();
} catch (IOException e) {
e.printStackTrace();
}
producer.close();
}
}
}
星火結構化流:
System.setProperty("hadoop.home.dir", "C:\\bigdata\\winutils");
final SparkSession sparkSession = SparkSession.builder().appName("Spark Data Processing").master("local[2]").getOrCreate();
// create kafka stream to get the lines
Dataset<Tuple2<String, String>> stream = sparkSession
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "test4")
.option("startingOffsets", "{\"test4\":{\"0\":-1,\"1\":-1}}")
.option("failOnDataLoss", "false")
.load().selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").as(Encoders.tuple(Encoders.STRING(), Encoders.STRING()));
Dataset<String> lines = stream.map((MapFunction<Tuple2<String, String>, String>) (Tuple2<String, String> tuple) -> tuple._2, Encoders.STRING());
Dataset<Row> result = lines.groupBy().count();
// Start running the query that prints the running counts to the console
StreamingQuery query = result//.orderBy("callTimeBin")
.writeStream()
.outputMode("complete")
.format("console")
.start();
// wait for the query to finish
try {
query.awaitTermination();
} catch (StreamingQueryException e) {
e.printStackTrace();
}
當我運行生產者發送的100行文件,查詢只返回51行。我讀了火花的調試日誌,並注意到如下內容:
17/02/15 10:52:49 DEBUG StreamExecution: Execution stats: ExecutionStats(Map(),List(),Map(watermark -> 1970-01-01T00:00:00.000Z))
17/02/15 10:52:49 DEBUG StreamExecution: Starting Trigger Calculation
17/02/15 10:52:49 DEBUG KafkaConsumer: Pausing partition test4-1
17/02/15 10:52:49 DEBUG KafkaConsumer: Pausing partition test4-0
17/02/15 10:52:49 DEBUG KafkaSource: Partitions assigned to consumer: [test4-1, test4-0]. Seeking to the end.
17/02/15 10:52:49 DEBUG KafkaConsumer: Seeking to end of partition test4-1
17/02/15 10:52:49 DEBUG KafkaConsumer: Seeking to end of partition test4-0
17/02/15 10:52:49 DEBUG Fetcher: Resetting offset for partition test4-1 to latest offset.
17/02/15 10:52:49 DEBUG Fetcher: **Fetched {timestamp=-1, offset=49} for partition test4-1
17/02/15 10:52:49 DEBUG Fetcher: Resetting offset for partition test4-1 to earliest offset.
17/02/15 10:52:49 DEBUG Fetcher: Fetched {timestamp=-1, offset=0} for partition test4-1**
17/02/15 10:52:49 DEBUG Fetcher: Resetting offset for partition test4-0 to latest offset.
17/02/15 10:52:49 DEBUG Fetcher: Fetched {timestamp=-1, offset=51} for partition test4-0
17/02/15 10:52:49 DEBUG KafkaSource: Got latest offsets for partition : Map(test4-1 -> 0, test4-0 -> 51)
17/02/15 10:52:49 DEBUG KafkaSource: GetOffset: ArrayBuffer((test4-0,51), (test4-1,0))
17/02/15 10:52:49 DEBUG StreamExecution: getOffset took 0 ms
17/02/15 10:52:49 DEBUG StreamExecution: triggerExecution took 0 ms
我不知道爲什麼test4-1總是重置爲ealiest偏移量。
如果有人知道如何從所有分區獲取所有消息,我將不勝感激。 謝謝,
謝謝,它現在使用0.10.0.1客戶端。 – taniGroup
我遇到了同樣的問題,並根據你的回答解決了它,謝謝。 – Mekal