_bootstrap_inner self.run() File "D:\Anaconda3\envs\py37\lib\site-packages\apache_beam\runners\worker\data_plane.py", ˓→ line 218, in run while not self._finished.wait(next_call - time.time()): File 7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 289, in _execute response = task() File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 362, in lambda: self.create_worker().do_instruction(request), request) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 607, in do_instruction getattr(request

_bootstrap_inner self.run() File "D:\Anaconda3\envs\py37\lib\site-packages\apache_beam\runners\worker\data_plane.py", ˓→ line 218, in run while not self._finished.wait(next_call - time.time()): File 7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 289, in _execute response = task() File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 362, in lambda: self.create_worker().do_instruction(request), request) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",␣ ˓→line 607, in do_instruction getattr(request

dataflow 1 2 3 6 5 4 ??? Vasiliki Kalavri | Boston University 2020 Dataflow worker activities worker 1 worker 2 worker 3 receive message deserialization processing serialization send message The DS2 model • Collect metrics per configurable observation window W • activity durations per worker • records processed Rprc and records pushed to output Rpsd 17 ??? Vasiliki Kalavri | Boston University The DS2 model • Collect metrics per configurable observation window W • activity durations per worker • records processed Rprc and records pushed to output Rpsd • Capture dependencies through the

driver to the worker. dstream.foreachRDD { rdd => val connection = createNewConnection() // executed at the driver rdd.foreach { record => connection.send(record) // executed at the worker } } 32 / driver to the worker. dstream.foreachRDD { rdd => val connection = createNewConnection() // executed at the driver rdd.foreach { record => connection.send(record) // executed at the worker } } 32 /

required • Key semantics preserved: values of the same key are always processed by the same worker • Popular keys cause imbalance w2 w1 w3 ??? Vasiliki Kalavri | Boston University 2020 Addressing University 2020 Dynamic resource allocation • Choose one among n workers • check the load of each worker and send the item to the least loaded one • load checking for every item can be expensive • Choose previously • Partial key grouping maps each key to both choices: the partitioner sends the item to the worker with the currently lowest load • no routing history required • state needs to be merged to produce

not be duplicates • Each worker maintains a Bloom Filter of all IDs it has seen: • if the filter returns false the record is not a duplicate • if it returns true, the worker sends a lookup to stable

according to the rate the consumer recycles buffers. Remote exchange: If tasks run on different worker nodes, the buffer can be recycled as soon as it is on the TCP channel. • If there is no buffer

University 2020 38 Safety • Avoid starvation: every data item is eventually processed • Ensure each worker is qualified: if load balancing is applied after fission, each instance must be capable of processing