教案大数据-学习lva1app

1、Apache Flink TrainingDataStream API BasicDecember 10, 2015DataStream APIStream ProcessingJava and ScalaAll examples here in Java for Flink 0.10Documentation available at 2DataStream API by Example3Window WordCount: main Method4public static void main(String args) throws Exception / set up the execut

2、ion environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / spl

3、it up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0” .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCoun

4、t Example);Stream Execution Environment5public static void main(String args) throws Exception / set up the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.Even

5、tTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0” .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tu

6、ple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCount Example);Data Sources6public static void main(String args) throws Exception / set up the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecu

7、tionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the

8、 tuple field 0 .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCount Example);Data types7public static void main(String args) throws Exception / set up the ex

9、ecution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) /

10、 split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0 .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordC

11、ount Example);Transformations8public static void main(String args) throws Exception / set up the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); Da

12、taStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0 .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1

13、 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCount Example);User functions9public static void main(String args) throws Exception / set up the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvir

14、onment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple fi

15、eld 0” .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCount Example);DataSinks10public static void main(String args) throws Exception / set up the execution

16、environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split u

17、p the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0” .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); / print result in command line counts.print(); / execute program env.execute(Socket WordCount Ex

18、ample);Execute!11public static void main(String args) throws Exception / set up the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); / configure event time env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStreamTuple

19、2 counts = env / read stream of words from socket .socketTextStream(localhost, 9999) / split up the lines in tuples containing: (word,1) .flatMap(new Splitter() / key stream by the tuple field 0” .keyBy(0) / compute counts every 5 minutes .timeWindow(Time.minutes(5) /sum up tuple field 1 .sum(1); /

20、print result in command line counts.print(); / execute program env.execute(Socket WordCount Example);Window WordCount: FlatMappublic static class Splitter implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) throws Exception / normalize and split t

21、he line String tokens = value.toLowerCase().split(W+); / emit the pairs for (String token : tokens) if (token.length() 0) out.collect( new Tuple2(token, 1); 12WordCount: Map: Interface13public static class Splitter implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, C

22、ollectorTuple2 out) throws Exception / normalize and split the line String tokens = value.toLowerCase().split(W+); / emit the pairs for (String token : tokens) if (token.length() 0) out.collect( new Tuple2(token, 1); WordCount: Map: Types14public static class Splitter implements FlatMapFunctionStrin

23、g, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) throws Exception / normalize and split the line String tokens = value.toLowerCase().split(W+); / emit the pairs for (String token : tokens) if (token.length() 0) out.collect( new Tuple2(token, 1); WordCount: Map: Collector15pu

24、blic static class Splitter implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) throws Exception / normalize and split the line String tokens = value.toLowerCase().split(W+); / emit the pairs for (String token : tokens) if (token.length() 0) out.co

25、llect( new Tuple2(token, 1); DataStream API Concepts16(Selected) Data TypesBasic Java TypesString, Long, Integer, Boolean,ArraysComposite TypesTuplesMany more (covered in the advanced slides)17TuplesThe easiest and most lightweight way of encapsulating data in FlinkTuple1 up to Tuple25Tuple2 person

26、= new Tuple2(Max, Mustermann”);Tuple3 person = new Tuple3(Max, Mustermann, 42);Tuple4 person = new Tuple4(Max, Mustermann, 42, true);/ zero based index!String firstName = person.f0;String secondName = person.f1;Integer age = person.f2;Boolean fired = person.f3;18Transformations: MapDataStream intege

27、rs = env.fromElements(1, 2, 3, 4);/ Regular Map - Takes one element and produces one elementDataStream doubleIntegers = integers.map(new MapFunction() Override public Integer map(Integer value) return value * 2; );doubleIntegers.print(); 2, 4, 6, 8/ Flat Map - Takes one element and produces zero, on

28、e, or more elements.DataStream doubleIntegers2 = integers.flatMap(new FlatMapFunction() Override public void flatMap(Integer value, Collector out) out.collect(value * 2); );doubleIntegers2.print(); 2, 4, 6, 819Transformations: Filter/ The DataStreamDataStream integers = env.fromElements(1, 2, 3, 4);

29、DataStream filtered = integers.filter(new FilterFunction() Override public boolean filter(Integer value) return value != 3; );filtered.print(); 1, 2, 420Transformations: KeyByA DataStream can be organized by a keyPartitions the data (all elements with the same key are processed by the same operator)

30、Certain operators are key-awareOperator state can be partitioned by key21/ (name, age) of employeesDataStreamTuple2 passengers = / group by second field (age)DataStream grouped = passengers.keyBy(1)Stephan, 18Fabian, 23Julia, 27Anna, 18Romeo, 27Anna, 18Stephan, 18Julia, 27Romeo, 27Fabian, 23Ben, 25B

31、en, 25Data Shipping StrategiesOptionally, you can specify how data is shipped between two transformationsForward: stream.forward()Only local communicationRebalance: stream.rebalance()Round-robin partitioningPartition by hash: stream.partitionByHash(.)Custom partitioning: stream.partitionCustom(.)Bro

32、adcast: stream.broadcast()Broadcast to all nodes22Rich Functions23RichFunctionsFunction interfaces have only one methodSingle abstract method (SAM)Support for Java8 Lambda functionsThere is a “Rich” variant for each function.RichFlatMapFunction, Additional methodsopen(Configuration c)close()getRunti

33、meContext()24RichFunctions & RuntimeContextRuntimeContext has useful methods:getIndexOfThisSubtask ()getNumberOfParallelSubtasks()getExecutionConfig() Hands out partitioned state (later discussed)getKeyValueState() 25Data Sources26Data Sources: CollectionsStreamExecutionEnvironment env = StreamExecu

34、tionEnvironment.getExecutionEnvironment();/ read from elementsDataStream names = env.fromElements(“Some”, “Example”, “Strings”);/ read from Java collectionList list = new ArrayList(); list.add(“Some”); list.add(“Example”); list.add(“Strings”);DataStream names = env.fromCollection(list);27Data Source

35、s: Files & SocketsStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/ read text socket from portDataStream socketLines = env.socketTextStream(localhost, 9999);/ read a text file ingesting new elements every 100 millisecondsDataStream localLines = env.readFileStrea

36、m(/path/to/file, 100, WatchType.PROCESS_ONLY_APPENDED);28Custom SourceFunctions & ConnectorsStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/ read data stream from custom source functionDataStreamTuple2 stream = env.addSource(new MySourceFunction();Flink has con

37、nectors for many stream serving systemsApache KafkaRabbitMQ29Data Sinks30Data SinksWrite as text file using toString()stream.writeAsText(“/path/to/file”)Write as CSV filestream.writeAsCsv(“/path/to/file”)Print to the standard outputstream.print()Emit to socketstream .writeToSocket(host,port,Serializ

38、ationSchema)31Custom Sinks & ConnectorsEmit data with a custom sink functionstream.addSink(new MySinkFunction();Several connectors Apache KafkaElasticsearchRolling Files (HDFS, S3, )32ExecutionPrograms are lazily executed when execute() is calledDataStream result;/ nothing happensresult.writeToSocke

39、t(.);/ nothing happensresult.writeAsText(/path/to/file, n, |);/ Execution really starts hereenv.execute();33Fault-Tolerance and Operator State34Fault Tolerance in FlinkFlink takes a checkpoint of an application every N milliseconds and rolls back to the checkpointed state in case of a failureEnable

40、exactly-once consistency (checkpoint every 5 seconds)env.enableCheckpointing(5000) Enable at-least-once consistency (for lower latency)env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE)Setting the interval to few seconds should be good for most applicationIf checkpointing is not enabled, no recovery guarantees are providedSee documentation for details: 35Stateful FunctionsAll DataStream functions can be statefulState is checkpointed and recovered in case of a failure (if checkpointing is enabled).You can define two types of stateLocal State: Functions can register local variabl