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 operators, straggler workers processing a tuple and all its derived results • Policy • each operator as a single-server queuing system • generalized Jackson networks • Action • predictive, at-once for all operators ??? Vasiliki processing a tuple and all its derived results • Policy • each operator as a single-server queuing system • generalized Jackson networks • Action • predictive, at-once for all operators Too fine-grained

distributed cut in a system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured captured A snapshot algorithm attempts to capture a coherent global state of a distributed system ??? Vasiliki Kalavri | Boston University 2020 Snapshotting Protocols p1 p2 p3 C m system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured

types • The system is unaware of which parts of an operator constitute state Streaming state 3 • Explicit state primitives including state types and interfaces • The system is aware of state persistent storage, e.g. a distributed filesystem or a database system • Available state backends in Flink: • In-memory • File system • RocksDB State backends 7 Vasiliki Kalavri | Boston University purposes! FsStateBackend • Stores state on TaskManager’s heap but checkpoints it to a remote file system • In-memory speed for local accesses and fault tolerance • Limited to TaskManager’s memory and might

Streamed? • Anything (if you write a serializer/deserializer for it) • Flink has a built-in type system which supports: • basic types, i.e., String, Long, Integer, Boolean, Array • composite types: 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

2 GB. If the size of an archive file is more than 2 GB, you could upload it to a distributed file system and then use the path in the command line option -pyarch. • Mix use of the above options You could pyflink-docs, Release release-1.15 1.1.1.5 Kubernetes Kubernetes is a popular container-orchestration system for automating computer application deployment, scaling, and management. This page shows you how management: User-defined function registration, dropping, listing, etc. • Executing SQL queries • Job configuration • Python dependency management • Job submission For more details of how to create a TableEnvironment

2 GB. If the size of an archive file is more than 2 GB, you could upload it to a distributed file system and then use the path in the command line option -pyarch. • Mix use of the above options You could pyflink-docs, Release release-1.16 1.1.1.5 Kubernetes Kubernetes is a popular container-orchestration system for automating computer application deployment, scaling, and management. This page shows you how management: User-defined function registration, dropping, listing, etc. • Executing SQL queries • Job configuration • Python dependency management • Job submission For more details of how to create a TableEnvironment

& reconfiguration ??? Vasiliki Kalavri | Boston University 2020 • To recover from failures, the system needs to • restart failed processes • restart the application and recover its state 2 Checkpointing and all required metadata, such as the application’s JAR file, into a remote persistent storage system • Zookeeper also holds state handles and checkpoint locations 5 JobManager failures ??? Vasiliki Vasiliki Kalavri | Boston University 2020 12 • Detect environment changes: external workload and system performance • Identify bottleneck operators, straggler workers, skew • Enumerate scaling actions

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 Vasiliki Kalavri | Boston University 2020 A distributed and fault-tolerant publish-subscribe messaging system and serves as the ingestion, storage, and messaging layer for large production streaming pipelines

latency constraints that can tolerate approximate results. Slow down the flow of data: • The system buffers excess data for later processing, once input rates stabilize. • Requires a persistent process of discarding data when input rates increase beyond system capacity. • Load shedding techniques operate in a dynamic fashion: the system detects an overload situation during runtime and selectively streams with known arrival rates C: system processing capacity H: headroom factor, i.e. a conservative estimate of the percentage of resources required by the system at steady state Load(N(I)): the load

....................................................................................... 22 4.14 System Resources....................................................................................... 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 reasons including the TaskManager (in case of a containerized setup), or by providing more TaskManagers. In general, a system already running under very high load during normal operations, will need much more time to catch-up