Boston University 2020 Control: When and how much to adapt? Mechanism: How to apply the re-configuration? 3 • Detect environment changes: external workload and system performance • Identify bottleneck Flink wordcount Every reconfiguration takes ~30s during which the system is unavailable Re-configuration requires state migration with correctness guarantees. ??? Vasiliki Kalavri | Boston University hidden from the application developer Live state migration ??? Vasiliki Kalavri | Boston University 2020 35 control command Helper operators, hidden from the application developer Live state

system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured A snapshot algorithm system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured A snapshot algorithm 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

environments to use. ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=1024m \ -Dyarn.application.name= \ -pyclientexec could not meet. ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=1024m \ -Dyarn.application.name= \ -Dyarn zip and word_count.py for the above example. As it executes the job on the JobManager in YARN application mode, the paths specified in -pyarch and -py are paths relative to shipfiles which is the directory

environments to use. ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=1024m \ -Dyarn.application.name= \ -pyclientexec could not meet. ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=1024m \ -Dyarn.application.name= \ -Dyarn zip and word_count.py for the above example. As it executes the job on the JobManager in YARN application mode, the paths specified in -pyarch and -py are paths relative to shipfiles which is the directory

factorValues)) } }) 17 Vasiliki Kalavri | Boston University 2020 Configuration options conf/flink-conf.yaml contains the configuration options as a collection of key-value pairs with format key:value Default parallelism for jobs. You can override this option by using env.setParallelism() in your application. taskmanager.numberOfTaskSlots: The number of parallel operator or user function instances that /bin/start-cluster.sh Stop Flink:
 ./bin/stop-cluster.sh Run an application with no arguments:
 ./bin/flink run ./examples/batch/WordCount.jar Run an application with input and output arguments:
 ./bin/flink run

• 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 sufficient number of processing slots in order to execute all tasks of an application. • The JobManager cannot restart the application until enough slots become available. • Restart is automatic if there standalone mode • The restart strategy determines how often the JobManager tries to restart the application and how long it waits between restart attempts. 4 TaskManager failures ??? Vasiliki Kalavri

and shrinks • queryable: Flink state can be queried via a REST API Rich Functions • open(Configuration c) • close() • getRuntimeContext() DataStream> input = … DataStream> { private ValueState averageState; @Override public void open (Configuration conf) { ValueStateDescriptor descriptor = new ValueStateDescriptor<>("moving String, String> { private ValueState blocked; @Override public void open(Configuration config) { blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean

handle object private var lastTempState: ValueState[Double] = _ override def open(parameters: Configuration): Unit = { // create state descriptor val lastTempDescriptor = new ValueStateDescriptor[Double]("lastTemp" rideState;
 private ValueState fareState;
 
 @Override
 public void open(Configuration config) { // initialize the state descriptors here
 rideState = getRuntimeContext()

output Redundancy elimination variations How can no-op or idempotent operators appear in an application? ??? Vasiliki Kalavri | Boston University 2020 23 Ensure the combination of A1, A2 is equivalent GET /dumprequest HTTP/1.1 Host: rve.org.uk Connection: keep-alive Accept: text/html,application/ xhtml+xml,application/ xml;q=0.9,*/*;q=0.8 User-Agent: Mozilla/5.0 (X11; Linux i686) AppleWebKit/537.22 GET /dumprequest HTTP/1.1 Host: rve.org.uk Connection: keep-alive Accept: text/html,application/ xhtml+xml,application/ xml;q=0.9,*/*;q=0.8 User-Agent: Mozilla/5.0 (X11; Linux i686) AppleWebKit/537.22

(framework or user code), as well as each network shuffle, takes time and adds to latency. 5. If the application emits through a transactional sink, the sink will only commit and publish transactions upon successful So far we have only looked at Flink-specific metrics. As long as latency & throughput of your application are in line with your expectations and it is checkpointing consistently, this is probably everything to the size of your application state (check the checkpointing metrics5 for an estimated size of the on-heap state). The possible reasons for growing state are very application-specific. Typically, an