教案大数据-学习lva1app

1、Apache Flink TrainingDataSet API BasicsJune 3rd, 2015DataSet APIBatch ProcessingJava, Scala, and PythonAll examples here in JavaMany concepts can be translated to the DataStream APIDocumentation available at 2DataSet API by Example3WordCount: main methodpublic static void main(String args) throws Ex

2、ception / set up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) in

3、putText.flatMap(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);4Execution Environmentpublic static void main(String args) throws Exception / set

4、up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMa

5、p(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);5Data Sourcespublic static void main(String args) throws Exception / set up the execution enviro

6、nment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / gro

7、up by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);6Data typespublic static void main(String args) throws Exception / set up the execution environment final ExecutionEnvi

8、ronment env = ExecutionEnvironment.getExecutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / group by the tuple field 0 .

9、groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);7Transformationspublic static void main(String args) throws Exception / set up the execution environment final ExecutionEnvironment env = Execut

10、ionEnvironment.getExecutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up t

11、uple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);8User functionspublic static void main(String args) throws Exception / set up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExe

12、cutionEnvironment(); / get input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceG

13、roup(new SumWords(); / emit result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);9DataSinkspublic static void main(String args) throws Exception / set up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); / get

14、 input data either from file or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emi

15、t result counts.writeAsCsv(args1, n, ); / execute program env.execute(WordCount Example);10Execute!public static void main(String args) throws Exception / set up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); / get input data either from fi

16、le or use example data DataSet inputText = env.readTextFile(args0); DataSetTuple2 counts = / split up the lines in tuples containing: (word,1) inputText.flatMap(new Tokenizer() / group by the tuple field 0 .groupBy(0) /sum up tuple field 1 .reduceGroup(new SumWords(); / emit result counts.writeAsCsv

17、(args1, n, ); / execute program env.execute(WordCount Example);11WordCount: Mappublic static class Tokenizer implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) / normalize and split the line String tokens = value.toLowerCase().split(W+); / emit t

18、he pairs for (String token : tokens) if (token.length() 0) out.collect( new Tuple2(token, 1); 12WordCount: Map: Interfacepublic static class Tokenizer implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) / normalize and split the line String tokens

19、 = value.toLowerCase().split(W+); / emit the pairs for (String token : tokens) if (token.length() 0) out.collect( new Tuple2(token, 1); 13WordCount: Map: Typespublic static class Tokenizer implements FlatMapFunctionString, Tuple2 Override public void flatMap(String value, CollectorTuple2 out) / norm

20、alize 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); 14WordCount: Map: Collectorpublic static class Tokenizer implements FlatMapFunctionString, Tuple2 Override public void flatMap

21、(String value, CollectorTuple2 out) / 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); 15WordCount: Reducepublic static class SumWords implements GroupReduceFunctionTuple2

22、, Tuple2 Override public void reduce(IterableTuple2 values, CollectorTuple2 out) int count = 0; String word = null; for (Tuple2 tuple : values) word = tuple.f0; count+; out.collect(new Tuple2(word, count); 16WordCount: Reduce: Interfacepublic static class SumWords implements GroupReduceFunctionTuple

23、2, Tuple2 Override public void reduce(IterableTuple2 values, CollectorTuple2 out) int count = 0; String word = null; for (Tuple2 tuple : values) word = tuple.f0; count+; out.collect(new Tuple2(word, count); 17WordCount: Reduce: Typespublic static class SumWords implements GroupReduceFunctionTuple2,

24、Tuple2 Override public void reduce(IterableTuple2 values, CollectorTuple2 out) int count = 0; String word = null; for (Tuple2 tuple : values) word = tuple.f0; count+; out.collect(new Tuple2(word, count); 18WordCount: Reduce: Collectorpublic static class SumWords implements GroupReduceFunctionTuple2,

25、 Tuple2 Override public void reduce(IterableTuple2 values, CollectorTuple2 out) int count = 0; String word = null; for (Tuple2 tuple : values) word = tuple.f0; count+; out.collect(new Tuple2(word, count); 19DataSet API Concepts20Data TypesBasic Java TypesString, Long, Integer, Boolean,ArraysComposit

26、e TypesTuplesMany more (covered in the advanced slides)21TuplesThe easiest and most lightweight way of encapsulating data in FlinkTuple1 up to Tuple25Tuple2 person = new Tuple2(Max, Mustermann”);Tuple3 person = new Tuple3(Max, Mustermann, 42);Tuple4 person = new Tuple4(Max, Mustermann, 42, true);/ z

27、ero based index!String firstName = person.f0;String secondName = person.f1;Integer age = person.f2;Boolean fired = person.f3;22Transformations: MapDataSet integers = env.fromElements(1, 2, 3, 4);/ Regular Map - Takes one element and produces one elementDataSet doubleIntegers = integers.map(new MapFu

28、nction() Override public Integer map(Integer value) return value * 2; );doubleIntegers.print(); 2, 4, 6, 8/ Flat Map - Takes one element and produces zero, one, or more elements.DataSet doubleIntegers2 = integers.flatMap(new FlatMapFunction() Override public void flatMap(Integer value, Collector out

29、) out.collect(value * 2); );doubleIntegers2.print(); 2, 4, 6, 823Transformations: Filter/ The DataSetDataSet integers = env.fromElements(1, 2, 3, 4);DataSet filtered = integers.filter(new FilterFunction() Override public boolean filter(Integer value) return value != 3; );integers.print(); 1, 2, 424G

30、roupings and Reduce/ (name, age) of employeesDataSetTuple2 employees = / group by second field (age)DataSet grouped = employees.groupBy(1) / return a list of age groups with its counts .reduceGroup(new CountSameAge();25NameAgeStephan18Fabian23Julia27Romeo27Anna18DataSets can be split into groupsGrou

31、ps are defined using a common keyAgeGroupCount182231272GroupReducepublic static class CountSameAge implements GroupReduceFunction Tuple2, Tuple2 Overridepublic void reduce(IterableTuple2 values, CollectorTuple2 out) Integer ageGroup = 0;Integer countsInGroup = 0;for (Tuple2 person : values) ageGroup

32、 = person.f1;countsInGroup+;out.collect(new Tuple2(ageGroup, countsInGroup);26Joining two DataSets/ authors (id, name, email)DataSetTuple3 authors = .;/ posts (title, content, author_id)DataSetTuple3 posts = .;DataSetTuple2Tuple3, Tuple3 archive = authors.join(posts).where(0).equalTo(2);27AuthorsIdN

33、ameemail1Fabianfabian.2Juliajulia.3Maxmax.4Romeoromeo.PostsTitleContentAuthor id2.4.4.1.2Joining two DataSets/ authors (id, name, email)DataSetTuple3 authors = .;/ posts (title, content, author_id)DataSetTuple3 posts = .;DataSetTuple2Tuple3, Tuple3 archive = authors.join(posts).where(0).equalTo(2);2

34、8ArchiveIdNameemailTitleContentAuthor id1Fabianfabian.12Juliajulia.22Juliajulia.23Romeoromeo.44Romeoromeo.4Join with join function/ authors (id, name, email)DataSetTuple3 authors = .;/ posts (title, content, author_id)DataSetTuple3 posts = .;/ (title, author name)DataSetTuple2 archive = authors.join

35、(posts).where(0).equalTo(2).with(new PostsByUser();public static class PostsByUser implements JoinFunctionTuple3, Tuple3, Tuple2 Overridepublic Tuple2 join( Tuple3 left, Tuple3 right) return new Tuple2(left.f1, right.f0);29ArchiveNameTitleFabian.Julia.Julia.Romeo.Romeo.Data SourcesTextreadTextFile(“

36、/path/to/file”)CSVreadCsvFile(“/path/to/file”)CollectionfromCollection(collection)fromElements(1,2,3,4,5)30Data Sources: CollectionsExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();/ read from elementsDataSet names = env.fromElements(“Some”, “Example”, “Strings”);/ read from

37、 Java collectionList list = new ArrayList(); list.add(“Some”); list.add(“Example”); list.add(“Strings”);DataSet names = env.fromCollection(list);31Data Sources: File-BasedExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();/ read text file from local or distributed file systemD

38、ataSet localLines = env.readTextFile(”/path/to/my/textfile);/ read a CSV file with three fieldsDataSetTuple3 csvInput = env.readCsvFile(“/the/CSV/file).types(Integer.class, String.class, Double.class);/ read a CSV file with five fields, taking only two of themDataSetTuple2 csvInput = env.readCsvFile(“/the/CSV/file) / take the firs