results: a local or instance variable that is accessed by a task’s business logic Operator state is scoped to an operator task, i.e. records processed by the same parallel task have access to the same state It cannot be accessed by other parallel tasks of the same or different operators Keyed state is scoped to a key defined in the operator’s input records • Flink maintains one state instance per key value key to the operator task that maintains the state for this key • State access is automatically scoped to the key of the current record so that all records with the same key access the same state State

30 ??? Vasiliki Kalavri | Boston University 2020 Pause-and-restart state migration • State is scoped to a single task • Each stateful task is responsible for processing and state management 31 31 ??? Vasiliki Kalavri | Boston University 2020 Pause-and-restart state migration • State is scoped to a single task • Each stateful task is responsible for processing and state management 31 operators ??? Vasiliki Kalavri | Boston University 2020 Pause-and-restart state migration • State is scoped to a single task • Each stateful task is responsible for processing and state management 31

each RDD using a given function. ▶ reduceByKey • Returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. ▶ countByValue • Returns a new DStream each RDD using a given function. ▶ reduceByKey • Returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. ▶ countByValue • Returns a new DStream each RDD using a given function. ▶ reduceByKey • Returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. ▶ countByValue • Returns a new DStream

us compute the statistical variance of this series? 3 • the sum of all the values • the sum of the squares of the values • the number of observations var = ∑ (xi − μ)2 N ??? Vasiliki Kalavri | us compute the statistical variance of this series? 3 • the sum of all the values • the sum of the squares of the values • the number of observations • μ = sum / count • var = (sum of squares / of this series? 3 • the sum of all the values • the sum of the squares of the values • the number of observations We can compute the three summary values in a single pass through the data. • μ =

Int): Real { TABLE state(tsum Int, cnt Int); INITIALIZE: { INSERT INTO state VALUES(Next, 1); } ITERATE: { UPDATE state SET tsum=tsum+Next Int): Real { TABLE state(tsum Int, cnt Int); INITIALIZE: { INSERT INTO state VALUES(Next, 1); } ITERATE: { UPDATE state SET tsum=tsum+Next Int): Real { TABLE state(tsum Int, cnt Int); INITIALIZE: { INSERT INTO state VALUES(Next, 1); } ITERATE: { UPDATE state SET tsum=tsum+Next, cnt=cnt+1;

maintain a hash table The more different elements we encounter in the stream, the more different hash values we shall see. Convert the stream into a multi-set of uniformly distributed random numbers using equal to: ??? Vasiliki Kalavri | Boston University 2020 The hash function h hashes x to any of N values with probability 1/N. Out of all x we hash: • around 50% will have a binary representation that so on… 6 ??? Vasiliki Kalavri | Boston University 2020 The hash function h hashes x to any of N values with probability 1/N. Out of all x we hash: • around 50% will have a binary representation that

EmitIntermediate(u, "1"); reduce(String key, Iterator values): // key: a URL // values: a list of counts int result = 0; for each v in values: result += ParseInt(v); Emit(key, AsString(result));

preserved: values of the same key might be routed to different workers • Workers are responsible for roughly the same amount of keys • No routing table is required • Key semantics preserved: values of Addressing skew • To address skew, the system needs to track the frequencies of the partitioning key values. • We can then use a hybrid partitioning function that treats normal keys and popular keys differently

. . . . . . . . . . . . . . . . . . . . . . . 31 1.3.6.1 Q1: ‘tuple’ object has no attribute ‘_values’ . . . . . . . . . . . . . . . . . . . . . 31 1.3.6.2 Q2: AttributeError: ‘int’ object has no attribute PyFlink in a clean environment. 1.3.6 Data type issues 1.3.6.1 Q1: ‘tuple’ object has no attribute ‘_values’ Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException: Error␣ ˓→received page) ˓→impl_fast.RowCoderImpl.encode_to_streamAttributeError: 'tuple' object has no attribute '_values' This issue is usually caused by the reason that it returns an object other than Row type in a Python

. . . . . . . . . . . . . . . . . . . . . . . 31 1.3.6.1 Q1: ‘tuple’ object has no attribute ‘_values’ . . . . . . . . . . . . . . . . . . . . . 31 1.3.6.2 Q2: AttributeError: ‘int’ object has no attribute PyFlink in a clean environment. 1.3.6 Data type issues 1.3.6.1 Q1: ‘tuple’ object has no attribute ‘_values’ Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException: Error␣ ˓→received page) ˓→impl_fast.RowCoderImpl.encode_to_streamAttributeError: 'tuple' object has no attribute '_values' This issue is usually caused by the reason that it returns an object other than Row type in a Python