2020 • To recover from failures, the system needs to • restart failed processes • restart the application and recover its state 2 Checkpointing guards the state from failures, but what about process tries to restart the application and how long it waits between restart attempts. 4 TaskManager failures ??? Vasiliki Kalavri | Boston University 2020 • The JobManager is a single point of failure Flink persistent storage system • Zookeeper also holds state handles and checkpoint locations 5 JobManager failures ??? Vasiliki Kalavri | Boston University 2020 When the JobManager fails all tasks are automatically

Boston University 2020 4 Distributed streaming systems will fail • how can we guard state against failures and guarantee correct results after recovery? • how can we ensure minimal downtime and fast order-preserving, reliable message transport, e.g. TCP. • Failures are single-node and fail- stop, i.e. no network partitions or multiple simultaneous failures are considered. • The secondary node uses keep- keep- alive requests to detect primary failures. 7 Vasiliki Kalavri | Boston University 2020 Recovery types 8 Vasiliki Kalavri | Boston University 2020 Recovery types • Precise recovery (exactly-once)

stored in the log. • For a topic with replication factor N, we will tolerate up to N-1 server failures without losing any records committed to the log. Vasiliki Kalavri | Boston University 2020 Resources

recovery) 5. Combine batch (historical) and stream processing 6. Ensure availability despite failures 7. Support distribution and automatic elasticity 8. Offer low-latency 7 2005 Vasiliki Kalavri

in memory • Use Spark's RDDs instead of replication • Parallel recovery mechanism in case of failures 44 input stream time-based micro-batches D-Streams • During an interval, input data received

brokers Where do stream processors read data from? 2 Challenges • can be distributed • out-of-sync sources may produce out-of-order streams • can be connected to the network • latency and unpredictable

every 5 minutes. // streaming DataFrame of schema {time: Timestamp, word: String} val calls = ... val actionHours = calls.groupBy(col("action"), window(col("time"), "1 hour", "5 minutes")) 64 / 79 Late Data

message queue, before it is processed by Apache Flink, which then writes the results to a database or calls a downstream system. In such a pipeline, latency can be introduced at each stage and for various

whereas if two 0s is the maximum we’ve seen, that indicates 4 distinct elements, … It takes 2r hash calls before we encounter a result with r 0s. 6 ??? Vasiliki Kalavri | Boston University 2020 Is this

p1 p2 p3 m m’ C’ A B ??? Vasiliki Kalavri | Boston University 2020 System model: • No failures during snapshotting • FIFO reliable channels: no lost or duplicate messages • Strongly connected