windows环境下flink入门实践操作示例

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.7.1</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>1.7.1</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.11</artifactId>
    <version>1.7.1</version>
</dependency>

package com.kl;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
public class SocketWindowWordCount {
   public static void main(String[] args) throws Exception {
      // the host and the port to connect to
      final String hostname;
      final int port;
      try {
         final ParameterTool params = ParameterTool.fromArgs(args);
         hostname = params.has("hostname") ? params.get("hostname") : "localhost";
         port = params.has("port") ? params.getInt("port"):9000;
      } catch (Exception e) {
         System.err.println("No port specified. Please run 'SocketWindowWordCount " +
            "--hostname--port', where hostname (localhost by default) " +
            "and port is the address of the text server");
         System.err.println("To start a simple text server, run 'netcat -l' and " +
            "type the input text into the command line");
         return;
      }
      // get the execution environment
      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      // get input data by connecting to the socket
      DataStreamtext = env.socketTextStream(hostname, port, "\n");
      // parse the data, group it, window it, and aggregate the counts
      DataStreamwindowCounts = text
            .flatMap(new FlatMapFunction() {
               public void flatMap(String value, Collectorout) {
                  for (String word : value.split("\\s")) {
                     out.collect(new WordWithCount(word, 1L));
                  } }})
            .keyBy("word")
            .timeWindow(Time.seconds(5))
            .reduce(new ReduceFunction() {
               public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                  return new WordWithCount(a.word, a.count + b.count);
               }});
      // print the results with a single thread, rather than in parallel
      windowCounts.print().setParallelism(1);
      env.execute("Socket Window WordCount");
   }
   /**
    * Data type for words with count.
    */
   public static class WordWithCount {
      public String word;
      public long count;
      public WordWithCount() {}
      public WordWithCount(String word, long count) {
         this.word = word;
         this.count = count;
      }
      @Override
      public String toString() {
         return word + " : " + count;
      }
   }
}

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>1.2.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>com.kl.SocketWindowWordCount</mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

FLINK RUN E:\FLINKWORKINGSPCE\FLINKDEMO\TARGET\FINLK-DEMO-1.0-SNAPSHOT.JAR --PORT 9000