Operator selectivity 6 • The number of output elements produced per number of input elements • a map operator has a selectivity of 1, i.e. it produces one output element for each input element it processes merge X merge A A X merge A1 merge A2 A2 A1 X X ??? Vasiliki Kalavri | Boston University 2020 map(String key, String value): // key: document name // value: document contents for each URL → list(v2) (k1, v1) → list(k2, v2) map() reduce() 25 ??? Vasiliki Kalavri | Boston University 2020 MapReduce combiners example: URL access frequency 26 map() reduce() GET /dumprequest HTTP/1

20 / 79 Transformations (2/4) ▶ map • Returns a new DStream by passing each element of the source DStream through a given function. ▶ flatMap • Similar to map, but each input item can be mapped to 21 / 79 Transformations (2/4) ▶ map • Returns a new DStream by passing each element of the source DStream through a given function. ▶ flatMap • Similar to map, but each input item can be mapped to 21 / 79 Transformations (2/4) ▶ map • Returns a new DStream by passing each element of the source DStream through a given function. ▶ flatMap • Similar to map, but each input item can be mapped to

Kalavri | Boston University 2020 Streaming word count textStream .flatMap {_.split("\\W+")} .map {(_, 1)} .keyBy(0) .sum(1) .print() “live and let live” “live” “and” “let” “live” (live nt
 val sensorData = env.addSource(new SensorSource)
 val maxTemp = sensorData
 .map(r => Reading(r.id,r.time,(r.temp-32)*(5.0/9.0)))
 .keyBy(_.id)
 .max("temp")
 maxTemp nt
 val sensorData = env.addSource(new SensorSource)
 val maxTemp = sensorData
 .map(r => Reading(r.id,r.time,(r.temp-32)*(5.0/9.0)))
 .keyBy(_.id)
 .max("temp")
 maxTemp

FIELD("data", DataTypes.STRING())])) def func(data: Row): return Row(data.id, data.data * 2) table.map(func).execute().print() +----+----------------------+--------------------------------+ | op | id | class MyFlatMapFunction(FlatMapFunction): def flat_map(self, value): for s in str(value.data).split('|'): yield Row(value.id, s) list(ds.flat_map(MyFlatMapFunction(), output_type=Types.ROW([Types.INT() part_size=1024 ** 3, rollover_interval=15 * 60 * 1000, inactivity_interval=5 *␣ ˓→60 * 1000)) .build()) ds.map(lambda i: (i[0] + 1, i[1]), Types.TUPLE([Types.INT(), Types.STRING()])).sink_ ˓→to(sink) # the result

FIELD("data", DataTypes.STRING())])) def func(data: Row): return Row(data.id, data.data * 2) table.map(func).execute().print() +----+----------------------+--------------------------------+ | op | id | class MyFlatMapFunction(FlatMapFunction): def flat_map(self, value): for s in str(value.data).split('|'): yield Row(value.id, s) list(ds.flat_map(MyFlatMapFunction(), output_type=Types.ROW([Types.INT() part_size=1024 ** 3, rollover_interval=15 * 60 * 1000, inactivity_interval=5 *␣ ˓→60 * 1000)) .build()) ds.map(lambda i: (i[0] + 1, i[1]), Types.TUPLE([Types.INT(), Types.STRING()])).sink_ ˓→to(sink) # the result

} } Map Function DataStream rides = env.addSource(new TaxiRideSource(...)); DataStream enrichedNYCRides = rides .filter(new RideCleansing.NYCFilter()) .map(new Enrichment()); class Enrichment implements MapFunction { @Override public EnrichedRide map(TaxiRide taxiRide) throws Exception { return new EnrichedRide(taxiRide); } } FlatMap Function DataStream> input = … DataStream> smoothed = input.keyBy(0).map(new Smoother()); public static class Smoother extends RichMapFunction, Tuple2

0 码力 | 45 页 | 3.00 MB | 1 年前

3

to count cardinalities up to 1 billion or 230 with an accuracy of 4%. • The hash value needs to map elements to M = log2(230) = 30 bits. Space requirements ??? Vasiliki Kalavri | Boston University to count cardinalities up to 1 billion or 230 with an accuracy of 4%. • The hash value needs to map elements to M = log2(230) = 30 bits. • We need 1024 counters, so m = 210 and we need p = log2m = 10 to count cardinalities up to 1 billion or 230 with an accuracy of 4%. • The hash value needs to map elements to M = log2(230) = 30 bits. • We need 1024 counters, so m = 210 and we need p = log2m = 10

val sensorData = env.addSource(new SensorSource) 
 val maxTemp = sensorData
 .map(r => Reading(r.id,r.time,(r.temp-32)*(5.0/9.0)))
 .keyBy(_.id) .timeWindow(Time.minutes(1)) Kalavri | Boston University 2020 val minTempPerWindow: DataStream[(String, Double)] = sensorData .map(r => (r.id, r.temperature)) .keyBy(_._1) .timeWindow(Time.seconds(15)) .reduce((r1, r2) Kalavri | Boston University 2020 val avgTempPerWindow: DataStream[(String, Double)] = sensorData .map(r => (r.id, r.temperature)) .keyBy(_._1) .timeWindow(Time.seconds(15)) .aggregate(new AvgTempFunction)

interface. What operations should state support? What state types can you think of? • Count, sum, list, map, … Vasiliki Kalavri | Boston University 2020 All data maintained by a task and used to compute results: List[T]) Flink’s state primitives 13 Vasiliki Kalavri | Boston University 2020 • MapState[K, V]: a map of keys and values • get(key: K), put(key: K, value: V), contains(key: K), remove(key: K) • iterators

selectivity 11 • Selectivity: how many records does the operator produce per record in its input? • map: 1 in 1 out • filter: 1 in, 1 or 0 out • flatMap, join: 1 in 0, 1, or more out • Cost: how many connected to multiple queries. 14 ??? Vasiliki Kalavri | Boston University 2020 Load Shedding Road Map (LSRM) • A pre-computed table that contains materialized load shedding plans ordered by how much