spawn new processes or release unused resources, safely terminate processes • Adjust dataflow channels and network connections • Re-partition and migrate state in a consistent manner • Block and temporarily block the affected dataflow subgraph • usually the operator to be scaled and upstream channels • All-at-once • move state to be migrated in one operation • high latency during migration single task • Each stateful task is responsible for processing and state management 31 block channels and upstream operators ??? Vasiliki Kalavri | Boston University 2020 Pause-and-restart state

spawn new processes or release unused resources, safely terminate processes • Adjust dataflow channels and network connections • Re-partition and migrate state in a consistent manner • Block and spawn new processes or release unused resources, safely terminate processes • Adjust dataflow channels and network connections • Re-partition and migrate state in a consistent manner • Block and spawn new processes or release unused resources, safely terminate processes • Adjust dataflow channels and network connections • Re-partition and migrate state in a consistent manner • Block and

Kalavri | Boston University 2020 System model: • No failures during snapshotting • FIFO reliable channels: no lost or duplicate messages • Strongly connected execution graph: each process can reach The initiator process: 1. Records its own state. 2. Sends a marker out on each of its outgoing channels. a. The marker is a special message that is not recorded in the snapshot but enforces the causal consistency. 3. Starts recording all data (application) messages it receives on all of its incoming channels. 20 ??? Vasiliki Kalavri | Boston University 2020 On receiving a marker (I) A process receiving

systems • Computations as Directed Acyclic Graphs (DAGs) • nodes are operators and edges are data channels • operators can accumulate state, have multiple inputs, express event- time custom window-based input port output port dataflow graph Dataflow graph • operators are nodes, data channels are edges • channels have FIFO semantics • streams of data elements flow continuously along edges Operators

University 2020 Revisiting the basics 4 Dataflow graph • operators are nodes, data channels are edges • channels have FIFO semantics • streams of data elements flow continuously along edges Operators

limited by the processing rate of the slowest task. • Parallel tasks are connected via virtual channels multiplexed over TCP connections: • In the presence of skew, a single overload channel can cause

Double) extends RichFlatMapFunction[Reading, (String, Double, Double)] { // the state handle object private var lastTempState: ValueState[Double] = _ override def open(parameters: Configuration): Unit = { Tuple2> {
 // define the state primitives here
 private ValueState rideState;
 private ValueState fareState;
 
 @Override
 public void open(Configuration ListCheckpointed { /** current offset for exactly once semantics */ private Long offset = 0L; private volatile boolean isRunning = true; @Override public void run(SourceContext

static class Smoother extends RichMapFunction, Tuple2> { private ValueState averageState; @Override public void open (Configuration conf) { public static class ControlFunction extends RichCoFlatMapFunction { private ValueState blocked; @Override public void open(Configuration config) { blocked window(TumblingEventTimeWindows.of(Time.minutes(1))) .reduce(new MyReducingMax(), new MyWindowFunction()); private static class MyReducingMax implements ReduceFunction { public SensorReading reduce(SensorReading

. . . . . . . . . . . . . . . 25 1.3.3.1 O1: InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not “opens java.lang” to unnamed import my_util 1.3.3 JDK issues 1.3.3.1 O1: InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not “opens java.lang” to un- un- named module @4e4aea35 : java.lang.reflect.InaccessibleObjectException: Unable to make field private final␣ ˓→byte[] java.lang.String.value accessible: module java.base does not "opens java.lang"␣

. . . . . . . . . . . . . . . 25 1.3.3.1 O1: InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not “opens java.lang” to unnamed import my_util 1.3.3 JDK issues 1.3.3.1 O1: InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not “opens java.lang” to un- un- named module @4e4aea35 : java.lang.reflect.InaccessibleObjectException: Unable to make field private final␣ ˓→byte[] java.lang.String.value accessible: module java.base does not "opens java.lang"␣