conda/lib/python3.7/site-packages/ ˓→pyflink/table/utils.py:55: FutureWarning: Schema passed to names= option, please pass␣ ˓→schema= explicitly. Will raise exception in future return pa.RecordBatch conda/lib/python3.7/site-packages/ ˓→pyflink/table/utils.py:55: FutureWarning: Schema passed to names= option, please pass␣ ˓→schema= explicitly. Will raise exception in future return pa.RecordBatch DataStream API. [7]: from pyflink.common import Row from pyflink.datastream import FlatMapFunction class MyFlatMapFunction(FlatMapFunction): def flat_map(self, value): for s in str(value.data).split('|'):

conda/lib/python3.7/site-packages/ ˓→pyflink/table/utils.py:55: FutureWarning: Schema passed to names= option, please pass␣ ˓→schema= explicitly. Will raise exception in future return pa.RecordBatch conda/lib/python3.7/site-packages/ ˓→pyflink/table/utils.py:55: FutureWarning: Schema passed to names= option, please pass␣ ˓→schema= explicitly. Will raise exception in future return pa.RecordBatch DataStream API. [7]: from pyflink.common import Row from pyflink.datastream import FlatMapFunction class MyFlatMapFunction(FlatMapFunction): def flat_map(self, value): for s in str(value.data).split('|'):

subscribing to a topic implicitly involves subscribing to all sub-topics of that topic, too. • Topic names are represented with URL-like notation and some systems also allow the use of wildcards. 21 Content-based

systems • Almost universally supported across streaming systems and languages albeit with various names and semantics • Allow un-blocking the processing of blocking operators by defining bounded portions

Examples Tuples Tuple1 through Tuple25 types. POJOs A POJO (plain old Java object) is any Java class that • has an empty default constructor • all fields are either ◦public, or ◦have a default getter Tuple2<>("Fred", 35); // zero based index! String name = person.f0; Integer age = person.f1; public class Person { public String name; public Integer age; public Person() {}; public total fare collected Lab 1 -- Ride Cleansing Transforming Data Transforming Data public static class EnrichedRide extends TaxiRide { public int startCell; public int endCell; public EnrichedRide()

Boston University 2020 DataStream API Basics Vasiliki Kalavri | Boston University 2020 case class Reading(id: String, time: Long, temp: Double)
 
 object MaxSensorReadings { def main(args: Array[String]) temperature”)
 }
 } Example: Sensor Readings 7 Vasiliki Kalavri | Boston University 2020 case class Reading(id: String, time: Long, temp: Double)
 
 object MaxSensorReadings { def main(args: Array[String]) temperature reading Example: Sensor Readings 8 Vasiliki Kalavri | Boston University 2020 case class Reading(id: String, time: Long, temp: Double)
 
 object MaxSensorReadings { def main(args: Array[String])

data types handled by the state are specified as Class or TypeInformation objects. 16 Registering state Vasiliki Kalavri | Boston University 2020 class TemperatureAlertFunction(val threshold: Double) assign name and get the state handle In the operator (FlatMap) class In the open() method Vasiliki Kalavri | Boston University 2020 class TemperatureAlertFunction(val threshold: Double) extends Rich flatMap(new MatchFunction());
 Java example 20 Vasiliki Kalavri | Boston University 2020 public static class EnrichmentFunction extends RichCoFlatMapFunction> {


average temperature per sensor. // The accumulator holds the sum of temperatures and an event count. class AvgTempFunction extends AggregateFunction [(String, Double), (String, Double, Int), (String, Double)] watermark. ProcessWindowFunction 18 Vasiliki Kalavri | Boston University 2020 public abstract class ProcessWindowFunction extends AbstractRichFunction { // Evaluates KEY key, Context ctx, Iterable vals, Collector out) throws Exception; public abstract class Context implements Serializable { // Returns the metadata of the window public

Announcements, updates, discussions • Website: vasia.github.io/dspa20 • Syllabus: /syllabus.html • Class schedule: /lectures.html • including today’s slides • Piazza: piazza.com/bu/spring2020/cs591k1/home applications 6 Vasiliki Kalavri | Boston University 2020 Grading Scheme (1) • No Exam • 5 in-class quizzes (10%): • Each quiz contributes 2% to the final grade • 3 hands-on assignments (40%): Kalavri | Boston University 2020 Schedule 9 vasia.github.io/dspa20/ lectures.html deadline no class guest lecture quizzes and announcements Vasiliki Kalavri | Boston University 2020 Guest Lectures

source: extend the Receiver class. ▶ Implement onStart() and onStop(). ▶ Call store(data) to store received data inside Spark. 16 / 79 Input Operations - Custom Sources (2/3) class CustomReceiver(host: String 79 Basic Operations ▶ Most of operations on DataFrame/Dataset are supported for streaming. case class Call(action: String, time: Timestamp, id: Int) val df: DataFrame = spark.readStream.json("s3://logs")