examples/TestTimer.java at master · xinyuiscool/examples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
packageorg.apache.beam.runners.samza.adapter;

importorg.apache.beam.runners.samza.SamzaPipelineOptions;
importorg.apache.beam.runners.samza.TestSamzaRunner;
importorg.apache.beam.sdk.Pipeline;
importorg.apache.beam.sdk.io.kafka.KafkaIO;
importorg.apache.beam.sdk.options.PipelineOptionsFactory;
importorg.apache.beam.sdk.transforms.Combine;
importorg.apache.beam.sdk.transforms.Count;
importorg.apache.beam.sdk.transforms.MapElements;
importorg.apache.beam.sdk.transforms.Values;
importorg.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
importorg.apache.beam.sdk.transforms.windowing.FixedWindows;
importorg.apache.beam.sdk.transforms.windowing.Window;
importorg.apache.beam.sdk.values.TypeDescriptors;
importorg.apache.kafka.common.serialization.StringDeserializer;
importorg.joda.time.Duration;
importorg.junit.Test;

/**
 * Created by xiliu on 2/5/18.
 */
publicclassTestTimer {

@Test
publicvoidtestTimer() {
SamzaPipelineOptionsoptions = PipelineOptionsFactory.as(SamzaPipelineOptions.class);
options.setRunner(TestSamzaRunner.class);
Pipelinepipeline = Pipeline.create(options);
pipeline
 .apply(KafkaIO.<String, String>read()
 .withTopic("TestTimerTopic")
 .withBootstrapServers("localhost:9092")
 .withKeyDeserializer(StringDeserializer.class)
 .withValueDeserializer(StringDeserializer.class)
 .withoutMetadata())
 .apply(Values.create())
 .apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(5)))
 .triggering(AfterProcessingTime.pastFirstElementInPane()
 .plusDelayOf(Duration.standardSeconds(5)))
 .discardingFiredPanes()
 .withAllowedLateness(Duration.standardSeconds(1)))
 .apply(Combine.globally(Count.<String>combineFn()).withoutDefaults())
 .apply(MapElements
 .into(TypeDescriptors.strings())
 .via(count -> {
Stringmsg = "Count in window is " + count;
System.out.println(msg);
returnmsg;
 }));

pipeline.run().waitUntilFinish();
 }

}