there are any problems, you could perform the following checks. Check the logging messages in the log file to see if there are any problems: # Get the installation directory of PyFlink python3 -c "import /path/to/python/site-packages/pyflink # Check the logging under the log directory ls -lh /path/to/python/site-packages/pyflink/log # You will see the log file as following: (continues on next page) 1.1. Getting previous page) # -rw-r--r-- 1 dianfu staff 45K 10 18 20:54 flink-dianfu-python-B-7174MD6R-1908. ˓→local.log Besides, you could also check if the files of the PyFlink package are consistent. It may happen that

there are any problems, you could perform the following checks. Check the logging messages in the log file to see if there are any problems: # Get the installation directory of PyFlink python3 -c "import /path/to/python/site-packages/pyflink # Check the logging under the log directory ls -lh /path/to/python/site-packages/pyflink/log # You will see the log file as following: (continues on next page) 1.1. Getting previous page) # -rw-r--r-- 1 dianfu staff 45K 10 18 20:54 flink-dianfu-python-B-7174MD6R-1908. ˓→local.log Besides, you could also check if the files of the PyFlink package are consistent. It may happen that

objects that have changed. • Logging to multiple systems • a Google Compute Engine instance can write logs to the monitoring system, to a database for later querying, and so on. • Data streaming from subscription's queue of messages. 25 Log-structured brokers Logs as message brokers • In typical message brokers, once a message is consumed it is deleted • Log-based message brokers take a different partitioned) log • A log is an append-only sequence of records on disk • a producer generates messages by simply appending them to the log and a consumer receives messages by reading the log sequentially

Kalavri | Boston University 2020 Let h be a hash function that maps each stream element into M = log2N bits, where N is the domain of input elements: For each element x, let rank(x) be the number maximum value of rank(.) seen so far. ̂n = 2R Claim: The maximum observed rank is a good estimate of log2n. In other words, the estimated number of distinct elements is equal to: ??? Vasiliki Kalavri | elements. We need a hash function that maps each input element to log2n bits. Then, each counter needs to be able to count up to log2(log2n) 0s. 13 ??? Vasiliki Kalavri | Boston University 2020 14 Combining

the Kafka cluster maintains a partitioned log. Each partition is an ordered, immutable sequence of records that is continually appended to—a structured commit log. An offset is a sequential id number time to free up disk space. 22 Vasiliki Kalavri | Boston University 2020 23 Partitions allow the log to scale beyond a size that will fit on a single server. Each individual partition must fit on will have a lower offset than M2 and appear earlier in the log. • A consumer instance sees records in the order they are stored in the log. • For a topic with replication factor N, we will tolerate

Kalavri | Boston University 2020 Using pseudocode (or the programming language of your choice), write a program that reads a stream of integers and computes: 29 1. the maximum number seen so far 2 control over arrival rate or order f’ ∞ ? Continuously arriving, possibly unbounded data f read write Complete data accessible in persistent storage 30 Vasiliki Kalavri | Boston University 2020 Consider The sensors monitor temperature and smoke levels and generate a measurement every 5 seconds. Write a program that every 1 minute emits the average temperature over the last 10 minutes. 31 Vasiliki

for some k, Lk = S we say that S is a pre-sequence of L and write S ⊆ L. If k < n, we say that S is a proper pre-sequence of L and write S ⊂ L. 24 Vasiliki Kalavri | Boston University 2020 Given University 2020 Pattern Queries with UDAs • UDAs process streams tuple-per-tuple • How can we write a UDA that detects a sequence of actions? • e.g. detect users who place an order, ask for a refund

University 2020 MemoryStateBackend • Stores state as regular objects on TaskManager’s heap • Low read/write latencies • OutOfMemoryError if large grows too large, GC pauses • Checkpoints sent to JobManager's and then scan one key at a time from that point (keys are sorted) • Merge: a lazy read-modify-write RocksDB 11 Vasiliki Kalavri | Boston University 2020 In conf/flink.conf.yaml: # Supported backends

identifiers of the last tuples they received. • In upstream backup, operators need to track and log tuple provenance / result lineage. Can such techniques be efficiently implemented? What if more

CDsBPHNRHML, UDHFGR 1RO6 mysOl_AHLlMF; FHRGSA.BMm/TDPTDPHBa/ElHLI-BCB-BMLLDBRMPs Flink 原生支持 Change Log Stream A C D E F G INSERT DELETE UPDATE INSERT DELETE UPDATE INSERT F3152 + Icebe7g