本地安装教程

    Flink runs on Linux, Mac OS X, and Windows. To be able to run Flink, the only requirement is to have a working Java 8.x installation. Windows users, please take a look at the Flink on Windows guide which describes how to run Flink on Windows for local setups.

    You can check the correct installation of Java by issuing the following command:

    If you have Java 8, the output will look something like this:

    2. Java(TM) SE Runtime Environment (build 1.8.0_111-b14)
    3. Java HotSpot(TM) 64-Bit Server VM (build 25.111-b14, mixed mode)
    • Download a binary from the . You can pickany Scala variant you like. For certain features you may also have to download one of the pre-bundled Hadoop jarsand place them into the /lib directory.
    • Go to the download directory.
    • Unpack the downloaded archive.
    1. $ cd ~/Downloads        # Go to download directory
    2. $ tar xzf flink-*.tgz   # Unpack the downloaded archive
    3. $ cd flink-1.9.0

    For MacOS X users, Flink can be installed through Homebrew.

    1. $ brew install apache-flink
    2. ...
    3. $ flink --version
    4. Version: 1.2.0, Commit ID: 1c659cf

    You can also verify that the system is running by checking the log files in the logs directory:

    1. $ tail log/flink-*-standalonesession-*.log
    2. INFO ... - Rest endpoint listening at localhost:8081
    3. INFO ... - http://localhost:8081 was granted leadership ...
    4. INFO ... - Web frontend listening at http://localhost:8081.
    5. INFO ... - Starting RPC endpoint for StandaloneResourceManager at akka://flink/user/resourcemanager .
    6. INFO ... - Starting RPC endpoint for StandaloneDispatcher at akka://flink/user/dispatcher .
    7. INFO ... - ResourceManager akka.tcp://flink@localhost:6123/user/resourcemanager was granted leadership ...
    8. INFO ... - Starting the SlotManager.
    9. INFO ... - Dispatcher akka.tcp://flink@localhost:6123/user/dispatcher was granted leadership ...
    10. INFO ... - Recovering all persisted jobs.
    11. INFO ... - Registering TaskManager ... at ResourceManager

    You can find the complete source code for this SocketWindowWordCount example in scala and on GitHub.

    1. object SocketWindowWordCount {
    3.     def main(args: Array[String]) : Unit = {
    5.         // the port to connect to
    6.         val port: Int = try {
    7.             ParameterTool.fromArgs(args).getInt("port")
    8.         } catch {
    9.             case e: Exception => {
    10.                 System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'")
    11.                 return
    12.             }
    13.         }
    15.         // get the execution environment
    16.         val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    18.         // get input data by connecting to the socket
    20.         // parse the data, group it, window it, and aggregate the counts
    21.         val windowCounts = text
    23.             .map { w => WordWithCount(w, 1) }
    24.             .keyBy("word")
    25.             .timeWindow(Time.seconds(5), Time.seconds(1))
    26.             .sum("count")
    28.         // print the results with a single thread, rather than in parallel
    29.         windowCounts.print().setParallelism(1)
    31.         env.execute("Socket Window WordCount")
    32.     }
    34.     // Data type for words with count
    35.     case class WordWithCount(word: String, count: Long)
    36. }
    1. public class SocketWindowWordCount {
    3.     public static void main(String[] args) throws Exception {
    5.         // the port to connect to
    6.         final int port;
    7.         try {
    8.             final ParameterTool params = ParameterTool.fromArgs(args);
    9.             port = params.getInt("port");
    10.         } catch (Exception e) {
    11.             System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'");
    12.             return;
    13.         }
    15.         // get the execution environment
    16.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    18.         // get input data by connecting to the socket
    19.         DataStream<String> text = env.socketTextStream("localhost", port, "\n");
    21.         // parse the data, group it, window it, and aggregate the counts
    22.         DataStream<WordWithCount> windowCounts = text
    23.             .flatMap(new FlatMapFunction<String, WordWithCount>() {
    24.                 @Override
    25.                 public void flatMap(String value, Collector<WordWithCount> out) {
    26.                     for (String word : value.split("\\s")) {
    27.                         out.collect(new WordWithCount(word, 1L));
    28.                     }
    29.                 }
    31.             .timeWindow(Time.seconds(5), Time.seconds(1))
    32.             .reduce(new ReduceFunction<WordWithCount>() {
    33.                 @Override
    34.                 public WordWithCount reduce(WordWithCount a, WordWithCount b) {
    35.                     return new WordWithCount(a.word, a.count + b.count);
    36.                 }
    37.             });
    39.         // print the results with a single thread, rather than in parallel
    40.         windowCounts.print().setParallelism(1);
    42.         env.execute("Socket Window WordCount");
    43.     }
    45.     // Data type for words with count
    46.     public static class WordWithCount {
    48.         public String word;
    49.         public long count;
    51.         public WordWithCount() {}
    53.         public WordWithCount(String word, long count) {
    54.             this.word = word;
    55.             this.count = count;
    56.         }
    58.         @Override
    59.         public String toString() {
    60.             return word + " : " + count;
    61.         }
    62.     }
    63. }

    Now, we are going to run this Flink application. It will read text froma socket and once every 5 seconds print the number of occurrences ofeach distinct word during the previous 5 seconds, i.e. a tumblingwindow of processing time, as long as words are floating in.

    • First of all, we use netcat to start local server via
    • Submit the Flink program:
    1. $ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000
    2. Starting execution of program

    Dispatcher: Overview (cont'd)

    • Words are counted in time windows of 5 seconds (processing time, tumblingwindows) and are printed to stdout. Monitor the TaskManager’s output fileand write some text in nc (input is sent to Flink line by line afterhitting ):
    1. $ nc -l 9000
    2. lorem ipsum
    3. ipsum ipsum ipsum
    4. bye

    The .out file will print the counts at the end of each time window as long as words are floating in, e.g.:

    1. $ tail -f log/flink-*-taskexecutor-*.out
    2. lorem : 1
    4. ipsum : 4

    To stop Flink when you’re done type: