东莞东城做网站公司/打开百度一下

processing-time

WallclockTimestampExtractor

event-time

ExtractRecordMetadataTimestamp子类

ingestion-time

ExtractRecordMetadataTimestamp子类

import org.apache.commons.io.FileUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.Stores;
import java.io.File;
import java.util.Locale;
import java.util.Properties;
/*** Demonstrates, using the low-level Processor APIs, how to implement the* WordCount program that computes a simple word occurrence histogram from an* input text.** In this example, the input stream reads from a topic named* "streams-file-input", where the values of messages represent lines of text;* and the histogram output is written to topic* "streams-wordcount-processor-output" where each record is an updated count of* a single word.** Before running this example you must create the input topic and the output* topic (e.g. via bin/kafka-topics.sh --create ...), and write some data to the* input topic (e.g. via bin/kafka-console-producer.sh). Otherwise you won't see* any data arriving in the output topic.*/
public class WordCountProcessorDemo {private static class MyProcessorSupplier implements ProcessorSupplier<String, String> {@Overridepublic Processor<String, String> get() {return new Processor<String, String>() {private ProcessorContext context;private KeyValueStore<String, Integer> kvStore;@Override@SuppressWarnings("unchecked")public void init(ProcessorContext context) {this.context = context;this.context.schedule(5000);this.kvStore = (KeyValueStore<String, Integer>) context.getStateStore("Counts");}@Overridepublic void process(String dummy, String line) {String[] words = line.toLowerCase(Locale.getDefault()).split(" ");System.out.println("line:" + line);for (String word : words) {Integer oldValue = this.kvStore.get(word);if (oldValue == null) {this.kvStore.put(word, 1);} else {this.kvStore.put(word, oldValue + 1);}}context.commit();}@Overridepublic void punctuate(long timestamp) {try (KeyValueIterator<String, Integer> iter = this.kvStore.all()) {System.out.println("----------- " + timestamp + "----------- ");//System.out.println(TimestampUtil.TimestampFormat(timestamp));while (iter.hasNext()) {KeyValue<String, Integer> entry = iter.next();System.out.println("[" + entry.key + ", " + entry.value + "]");context.forward(entry.key, entry.value.toString());}}}@Overridepublic void close() {}};}}public static void main(String[] args) throws Exception {Properties props = new Properties();props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount-processor");props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "breath:9092");props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG,"org.apache.kafka.streams.processor.WallclockTimestampExtractor");TopologyBuilder builder = new TopologyBuilder();builder.addSource("Source", "streams-file-input");builder.addProcessor("Process", new MyProcessorSupplier(), "Source");builder.addStateStore(Stores.create("Counts").withStringKeys().withIntegerValues().inMemory().build(),"Process");builder.addSink("Sink", "streams-wordcount-processor-output", "Process");KafkaStreams streams = new KafkaStreams(builder, props);streams.start();Thread.sleep(20000L);streams.close();// 清空state storeString appStateDir = props.get(StreamsConfig.STATE_DIR_CONFIG) + System.getProperty("file.separator")+ props.get(StreamsConfig.APPLICATION_ID_CONFIG);FileUtils.deleteDirectory(new File(appStateDir));// 清空application相关中间topic以及input topic的offsetString kafkaHome = System.getenv("KAFKA_HOME");Runtime runtime = Runtime.getRuntime();runtime.exec(kafkaHome + "/bin/kafka-streams-application-reset.sh " + "--application-id "+ props.get(StreamsConfig.APPLICATION_ID_CONFIG) + " " + "--bootstrap-servers breath:9092 "+ "--zookeeper breath:2181/kafka01021 " + "--input-topics streams-file-input");}
}