flink

ac2be7e7 · liugen.wen · ac2be7e7 · ac2be7e7 · ac2be7e7
--- a/.gitignore
+++ b/.gitignore
+## Java ##
+
+*.class
+*.jar
+*.war
+*.ear
+target/
+build/
+
+## Eclipse ##
+
+*.pydevproject
+.metadata
+.gradle
+bin/
+tmp/
+*.tmp
+*.bak
+*.swp
+*~.nib
+local.properties
+.settings/
+.loadpath
+.Rhistory
+.classpath
+.idea/
+.project
+pom.xml
+karma-*/pom.xml
+
+# External tool builders
+.externalToolBuilders/
+
+# Locally stored "Eclipse launch configurations"
+*.launch
+
+# CDT-specific
+.cproject
+
+# PDT-specific
+.buildpath
+
+# sbteclipse plugin
+.target
+
+# TeXlipse plugin
+.texlipse
+
+
+## JetBrains ##
+
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm
+
+## Directory-based project format
+.idea/
+# if you remove the above rule, at least ignore user-specific stuff:
+# .idea/workspace.xml
+# .idea/tasks.xml
+# and these sensitive or high-churn files:
+# .idea/dataSources.ids
+# .idea/dataSources.xml
+# .idea/sqlDataSources.xml
+# .idea/dynamic.xml
+
+## File-based project format
+*.ipr
+*.iws
+*.iml
+
+## Additional for IntelliJ
+out/
+
+# generated by mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# generated by JIRA plugin
+atlassian-ide-plugin.xml
+
+# generated by Crashlytics plugin (for Android Studio and Intellij)
+com_crashlytics_export_strings.xml
+*.pyc
+*.DS_Store
+
--- a/pom.xml
+++ b/pom.xml
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>flinktest</groupId>
+    <artifactId>flinktest</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_2.11</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-java</artifactId>
+            <version>1.7.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_2.12</artifactId>
+            <version>1.7.1</version>
+        </dependency>
+    </dependencies>
+
+
+    <build>
+        <!--<resources>-->
+            <!--<resource>-->
+                <!--<directory>src/main/resources</directory>//这个地方是把resources下的配置文件打进包里-->
+            <!--</resource>-->
+        <!--</resources>-->
+        <sourceDirectory>src/main/java</sourceDirectory>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                    <encoding>UTF-8</encoding>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>2.3</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <transformers>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass>demo.Main</mainClass>
+                                </transformer>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>reference.conf</resource>
+                                </transformer>
+
+                            </transformers>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+
+
+        </plugins>
+    </build>
+
+
+
+</project>
\ No newline at end of file
--- a/src/main/java/demo/Main.java
+++ b/src/main/java/demo/Main.java
+package demo;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.util.Collector;
+
+/**
+ * @author Laurence Cao
+ *
+ */
+public class Main {
+
+    public static void main(String[] args) throws Exception {
+        // Checking input parameters
+        final ParameterTool params = ParameterTool.fromArgs(args);
+
+        // set up the execution environment
+        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        // make parameters available in the web interface
+        env.getConfig().setGlobalJobParameters(params);
+
+        // get input data
+        DataStream<String> text;
+        if (params.has("input")) {
+            // read the text file from given input path
+            text = env.readTextFile(params.get("input"));
+        } else {
+            System.out.println("Executing WordCount example with default input data set.");
+            System.out.println("Use --input to specify file input.");
+            // get default test text data
+            text = env.fromElements(WordCountData.WORDS);
+        }
+
+        DataStream<Tuple2<String, Integer>> counts =
+            // split up the lines in pairs (2-tuples) containing: (word,1)
+            text.flatMap(new Tokenizer())
+                // group by the tuple field "0" and sum up tuple field "1"
+                .keyBy(0).sum(1);
+
+        // emit result
+        if (params.has("output")) {
+            counts.writeAsText(params.get("output"));
+        } else {
+            System.out.println("Printing result to stdout. Use --output to specify output path.");
+            counts.print();
+        }
+
+        // execute program
+        env.execute("Streaming WordCount");
+    }
+
+    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
+
+        private static final long serialVersionUID = 1L;
+
+      //  @Override
+        public void flatMap(String value, Collector<Tuple2<String, Integer>> 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));
+                }
+            }
+        }
+    }
+
+}
+
+class WordCountData {
+    public static final String[] WORDS = new String[] {"To be, or not to be,--that is the question:--",
+        "Whether 'tis nobler in the mind to suffer", "The slings and arrows of outrageous fortune",
+        "Or to take arms against a sea of troubles,", "And by opposing end them?--To die,--to sleep,--",
+        "No more; and by a sleep to say we end", "The heartache, and the thousand natural shocks",
+        "That flesh is heir to,--'tis a consummation", "Devoutly to be wish'd. To die,--to sleep;--",
+        "To sleep! perchance to dream:--ay, there's the rub;", "For in that sleep of death what dreams may come,",
+        "When we have shuffled off this mortal coil,", "Must give us pause: there's the respect",
+        "That makes calamity of so long life;", "For who would bear the whips and scorns of time,",
+        "The oppressor's wrong, the proud man's contumely,", "The pangs of despis'd love, the law's delay,",
+        "The insolence of office, and the spurns", "That patient merit of the unworthy takes,",
+        "When he himself might his quietus make", "With a bare bodkin? who would these fardels bear,",
+        "To grunt and sweat under a weary life,", "But that the dread of something after death,--",
+        "The undiscover'd country, from whose bourn", "No traveller returns,--puzzles the will,",
+        "And makes us rather bear those ills we have", "Than fly to others that we know not of?",
+        "Thus conscience does make cowards of us all;", "And thus the native hue of resolution",
+        "Is sicklied o'er with the pale cast of thought;", "And enterprises of great pith and moment,",
+        "With this regard, their currents turn awry,", "And lose the name of action.--Soft you now!",
+        "The fair Ophelia!--Nymph, in thy orisons", "Be all my sins remember'd."};
+}
\ No newline at end of file