logical conjunction • if A is a projection on multiple attributes • if A is an idempotent aggregation Operator separation A A2 A1 Separate operators into smaller computational steps • beneficial Boston University 2020 24 • Cost of Merge = 0.5 • Cost of A = 0.5 • Splitting A allows a pre-aggregation similar to what combiners do in MapReduce Operator separation merge X merge A A X merge micro-batches D-Streams • During an interval, input data received is stored using RDDs • A D-Stream is a group of such RDDs which can be processed using common operators 45 Example • pageViews is a D-Stream

Kinesis, ... TwitterUtils.createStream(ssc, None) KafkaUtils.createStream(ssc, [ZK quorum], [consumer group id], [number of partitions]) 15 / 79 Input Operations - Custom Sources (1/3) ▶ To create a custom express it as the following SQL query. SELECT action, WINDOW(time, "1 hour"), COUNT * FROM events GROUP BY action, WINDOW(time, "1 hour") 61 / 79 Structured Streaming Example (3/3) val inputDF = spark where("id > 10") // using untyped APIs ds.filter(_.id > 10).map(_.action) // using typed APIs // Aggregation df.groupBy("action") // using untyped API ds.groupByKey(_.action) // using typed API // SQL commands

val sensorData: DataStream[SensorReading] = ... val avgTemp = sensorData .keyBy(_.id) // group readings in 1s event-time windows .window(TumblingEventTimeWindows.of(Time.seconds(1))) .process(new functions define the computation that is performed on the elements of a window • Incremental aggregation functions are applied when an element is added to a window: • They maintain a single value as

measurements analysis • Monitoring applications • Complex filtering and alarm activation • Aggregation of multiple sensors and joins • Examples • Real-time statistics, e.g. weather maps • Monitor activity analysis • Visualization and aggregation • impressions, clicks, transactions, likes, comments • Analytics on user activity • Filtering, aggregation, joins with static data (e.g. user profile

given the constraint that system throughput matches the data input rate • In the case of known aggregation functions, results can be scaled using approximate query processing techniques, where accuracy a data stream manager. (VLDB ’03) • N. Tatbul and S. Zdonik. Window-aware load shedding for aggregation queries over data streams. (VLDB’06) • N. Tatbul, U. Çetintemel, and S. Zdonik. Staying fit:

Operator replicates a stream, commonly to be used as input to multiple downstream operators. • Group by / Partition Operators split a stream into sub-streams according to a function or the event contents Kalavri | Boston University 2020 CQL GroupBy Example Select IStream(Count(*)) From S1 [Rows 1000] Group By S1.B Count the number or events in the last 1000 rows for each value of B 20 Vasiliki Kalavri University 2020 Some queries expressed using aggregates are monotonic: SELECT DeptNo FROM empl GROUP BY DeptNo HAVING SUM(empl.Sal) > 10000 The introduction of a new empl can only expand the set

Consumers label themselves with a consumer group name, and each record published to a topic is delivered to one consumer instance within each subscribing consumer group. Consumer instances can be in separate separate processes or on separate machines. If all the consumer instances have the same consumer group, then the records will effectively be load balanced over the consumer instances. If all the consumer

topic T can be viewed as becoming a member of a group T. • Publishing an event on topic T can be viewed as broadcasting the event to all members of group T. • Topic hierarchies allow topic organization

as ranges • On restore, reads are sequential within each key-group, and often across multiple key-groups • The metadata of key-group-to-subtask assignments are small. No need to maintain explicit

---------------------------- Stream Mode: 12:01> SELECT Name, SUM(Score), MAX(Time) FROM USER_SCORES GROUP BY Name; Flink 在阿里的服务情况 集群规模 超万台 状态数据 PetaBytes 事件处理 十万亿/天 峰值能力 17亿/秒 Flink 的过去 offline Real-time