Learning Apache Flink

• Learning Apache Flink

• By: Tanmay Deshpande

• Publisher: Packt Publishing

• Pub. Date: February 20, 2017

• Web ISBN-13: 978-1-78646-726-3

• Print ISBN-13: 978-1-78646-622-8

• Pages in Print Edition: 280

• Subscriber Rating: [0 Ratings]

  
  

1. Map
   
2. 
   
3. This is one of the simplest transformations, where the input is one data stream and the output is also one data stream.
   
4. 
   
5. In Java:
   
6. 
   
7. inputStream.map(new MapFunction<Integer, Integer>() {
   
8.   @Override
   
9.   public Integer map(Integer value) throws Exception {
   
10.         return 5 * value;
    
11.       }
    
12.     });
    
13. In Scala:
    
14. 
    
15. inputStream.map { x => x * 5

复制代码

1. FlatMap
   
2. 
   
3. FlatMap takes one record and outputs zero, one, or more than one record.
   
4. 
   
5. In Java:
   
6. 
   
7. inputStream.flatMap(new FlatMapFunction<String, String>() {
   
8.     @Override
   
9.     public void flatMap(String value, Collector<String> out)
   
10.         throws Exception {
    
11.         for(String word: value.split(" ")){
    
12.             out.collect(word);
    
13.         }
    
14.     }
    
15. });
    
16. In Scala:
    
17. 
    
18. inputStream.flatMap { str => str.split(" ") }

复制代码

1. Filter
   
2. 
   
3. Filter functions evaluate the conditions and then, if they result as true, only emit the record. Filter functions can output zero records.
   
4. 
   
5. In Java:
   
6. 
   
7. inputStream.filter(new FilterFunction<Integer>() {
   
8.     @Override
   
9.     public boolean filter(Integer value) throws Exception {
   
10.         return value != 1;
    
11.     }
    
12. });
    
13. In Scala:
    
14. 
    
15. inputStream.filter { _ != 1 }

复制代码

1. KeyBy
   
2. 
   
3. KeyBy logically partitions the stream-based on the key. Internally it uses hash functions to partition the stream. It returns KeyedDataStream.
   
4. 
   
5. In Java:
   
6. 
   
7. inputStream.keyBy("someKey");
   
8. In Scala:
   
9. 
   
10. inputStream.keyBy("someKey")

复制代码

1. Reduce
   
2. 
   
3. Reduce rolls out the KeyedDataStream by reducing the last reduced value with the current value. The following code does the sum reduce of a KeyedDataStream.
   
4. 
   
5. In Java:
   
6. 
   
7. keyedInputStream. reduce(new ReduceFunction<Integer>() {
   
8.     @Override
   
9.     public Integer reduce(Integer value1, Integer value2)
   
10.     throws Exception {
    
11.         return value1 + value2;
    
12.     }
    
13. });
    
14. In Scala:
    
15. 
    
16. keyedInputStream. reduce { _ + _ }

复制代码

1. Fold
   
2. 
   
3. Fold rolls out the KeyedDataStream by combining the last folder stream with the current record. It emits a data stream back.
   
4. 
   
5. In Java:
   
6. 
   
7. keyedInputStream keyedStream.fold("Start", new FoldFunction<Integer, String>() {
   
8.     @Override
   
9.     public String fold(String current, Integer value) {
   
10.         return current + "=" + value;
    
11.     }
    
12.   });
    
13. In Scala:
    
14. 
    
15. keyedInputStream.fold("Start")((str, i) => { str + "=" + i })
    
16. The preceding given function when applied on a stream of (1,2,3,4,5) would emit a stream like this: Start=1=2=3=4=5

复制代码

1. package com.demo.chapter06
   
2. 
   
3. import org.apache.flink.api.scala._
   
4. import org.apache.flink.ml.math.Vector
   
5. import org.apache.flink.ml.common.LabeledVector
   
6. import org.apache.flink.ml.classification.SVM
   
7. import org.apache.flink.ml.RichExecutionEnvironment
   
8. 
   
9. object MySVMApp {
   
10.   def main(args: Array[String]) {
    
11.     // set up the execution environment
    
12.     val pathToTrainingFile: String = "iris-train.txt"
    
13.     val pathToTestingFile: String = "iris-train.txt"
    
14.     val env = ExecutionEnvironment.getExecutionEnvironment
    
15. 
    
16.     // Read the training dataset, from a LibSVM formatted file
    
17.     val trainingDS: DataSet[LabeledVector] =
    
18.     env.readLibSVM(pathToTrainingFile)
    
19. 
    
20.     // Create the SVM learner
    
21.     val svm = SVM()
    
22.       .setBlocks(10)
    
23. 
    
24.     // Learn the SVM model
    
25.     svm.fit(trainingDS)
    
26. 
    
27.     // Read the testing dataset
    
28.     val testingDS: DataSet[Vector] =
    
29.     env.readLibSVM(pathToTestingFile).map(_.vector)
    
30. 
    
31.     // Calculate the predictions for the testing dataset
    
32.     val predictionDS: DataSet[(Vector, Double)] =
    
33.     svm.predict(testingDS)
    
34.     predictionDS.writeAsText("out")
    
35. 
    
36.     env.execute("Flink SVM App")
    
37.   }
    
38. }

复制代码

1. #!/usr/bin/env python
   
2. 
   
3. """
   
4. Convert CSV file to libsvm format. Works only with numeric variables.
   
5. Put -1 as label index (argv[3]) if there are no labels in your file.
   
6. Expecting no headers. If present, headers can be skipped with argv[4] == 1.
   
7. 
   
8. """
   
9. 
   
10. import sys
    
11. import csv
    
12. from collections import defaultdict
    
13. 
    
14. def construct_line( label, line ):
    
15.         new_line = []
    
16.         if float( label ) == 0.0:
    
17.                 label = "0"
    
18.         new_line.append( label )
    
19. 
    
20.         for i, item in enumerate( line ):
    
21.                 if item == '' or float( item ) == 0.0:
    
22.                         continue
    
23.                 new_item = "%s:%s" % ( i + 1, item )
    
24.                 new_line.append( new_item )
    
25.         new_line = " ".join( new_line )
    
26.         new_line += "\n"
    
27.         return new_line
    
28. 
    
29. # ---
    
30. 
    
31. input_file = sys.argv[1]
    
32. output_file = sys.argv[2]
    
33. 
    
34. try:
    
35.         label_index = int( sys.argv[3] )
    
36. except IndexError:
    
37.         label_index = 0
    
38. 
    
39. try:
    
40.         skip_headers = sys.argv[4]
    
41. except IndexError:
    
42.         skip_headers = 0
    
43. 
    
44. i = open( input_file, 'rb' )
    
45. o = open( output_file, 'wb' )
    
46. 
    
47. reader = csv.reader( i )
    
48. 
    
49. if skip_headers:
    
50.         headers = reader.next()
    
51. 
    
52. for line in reader:
    
53.         if label_index == -1:
    
54.                 label = '1'
    
55.         else:
    
56.                 label = line.pop( label_index )
    
57. 
    
58.         new_line = construct_line( label, line )
    
59.         o.write( new_line )

复制代码