-
Notifications
You must be signed in to change notification settings - Fork 60
/
WorkingTimeMonitor.java
150 lines (130 loc) · 6.17 KB
/
WorkingTimeMonitor.java
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
/*
* Copyright 2018 data Artisans GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.dataartisans.examples.infoworld.eventdrivenapp;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import com.dataartisans.examples.infoworld.utils.TaxiRide;
import com.dataartisans.examples.infoworld.utils.TaxiRides;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
/**
* This example application ingests a stream of taxi ride events and notifies taxi drivers when
* they have to stop working according to work time regulations.
*
* A driver's shift may last for at most 12 hours, meaning that rides may only be started within a
* 12 hour period that begins with the first ride of a shift.
* After a 12 hour shift, a driver needs to take a break of at least 8 hours.
*
* The application notifies drivers until when the last ride of their shift may be started and
* when they violate the regulation.
*
* The stream of taxi ride events is read from a gzipped CSV file.
* The file can be downloaded from
*
* https://drive.google.com/file/d/0B0TBL8JNn3JgTGNJTEJaQmFMbk0
*
* When starting the application the path to the file has to be provided as a parameter.
*/
public class WorkingTimeMonitor {
public static void main(String[] args) throws Exception {
// check parameter
if (args.length != 1) {
System.err.println("Please provide the path to the taxi rides file as a parameter");
}
String inputPath = args[0];
// create execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// configure event-time and watermarks
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.getConfig().setAutoWatermarkInterval(1000L);
// ingest stream of taxi rides.
DataStream<TaxiRide> rides = TaxiRides.getRides(env, inputPath);
DataStream<Tuple2<String, String>> notifications = rides
// partition stream by the driver's license id
.keyBy(r -> r.licenseId)
// monitor ride events and generate notifications
.process(new MonitorWorkTime());
// print notifications
notifications.print();
// run the application
env.execute();
}
/**
* A KeyedProcessFunction that monitors taxi rides for regulation compliance.
*
* The function emits a notification for drivers.
*/
public static class MonitorWorkTime extends KeyedProcessFunction<String, TaxiRide, Tuple2<String, String>> {
// time constants in milliseconds
private static final long ALLOWED_WORK_TIME = 12 * 60 * 60 * 1000; // 12 hours
private static final long REQ_BREAK_TIME = 8 * 60 * 60 * 1000; // 8 hours
private static final long CLEAN_UP_INTERVAL = 24 * 60 * 60 * 1000; // 24 hours
private transient DateTimeFormatter formatter;
// state handle to store the starting time of a shift
ValueState<Long> shiftStart;
@Override
public void open(Configuration conf) {
// register state handle
shiftStart = getRuntimeContext().getState(
new ValueStateDescriptor<>("shiftStart", Types.LONG));
// initialize time formatter
this.formatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
}
@Override
public void processElement(
TaxiRide ride,
Context ctx,
Collector<Tuple2<String, String>> out) throws Exception {
// look up start time of the last shift
Long startTs = shiftStart.value();
if (startTs == null ||
startTs < ride.pickUpTime - (ALLOWED_WORK_TIME + REQ_BREAK_TIME)) {
// this is the first ride of a new shift.
startTs = ride.pickUpTime;
shiftStart.update(startTs);
long endTs = startTs + ALLOWED_WORK_TIME;
out.collect(Tuple2.of(ride.licenseId,
"New shift started. Shift ends at " + formatter.print(endTs) + "."));
// register timer to clean up the state in 24h
ctx.timerService().registerEventTimeTimer(startTs + CLEAN_UP_INTERVAL);
} else if (startTs < ride.pickUpTime - ALLOWED_WORK_TIME) {
// this ride started after the allowed work time ended.
// it is a violation of the regulations!
out.collect(Tuple2.of(ride.licenseId,
"This ride violated the working time regulations."));
}
}
@Override
public void onTimer(
long timerTs,
OnTimerContext ctx,
Collector<Tuple2<String, String>> out) throws Exception {
// remove the shift state if no new shift was started already.
Long startTs = shiftStart.value();
if (startTs == timerTs - CLEAN_UP_INTERVAL) {
shiftStart.clear();
}
}
}
}