first commit

Author: zhangblue

.gitignore

@@ -0,0 +1,32 @@
+# Created by .ignore support plugin (hsz.mobi)
+### Scala template
+*.class
+*.log
+
+### Java template
+# Compiled class file
+*.class
+
+# Log file
+*.log
+
+# BlueJ files
+*.ctxt
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+*.iml
+.idea/
+target/

pom.xml

@@ -0,0 +1,62 @@
+<?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>com.zhangblue</groupId>
+  <artifactId>FlinkTutorial</artifactId>
+  <version>1.0-SNAPSHOT</version>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-scala_2.11</artifactId>
+      <version>1.7.2</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-streaming-scala_2.11</artifactId>
+      <version>1.7.2</version>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <!--该插件用于将scala代码编译成class文件-->
+      <plugin>
+        <groupId>net.alchim31.maven</groupId>
+        <artifactId>scala-maven-plugin</artifactId>
+        <version>3.4.6</version>
+        <executions>
+          <execution>
+            <!--声明绑定到maven的compile阶段-->
+            <goals>
+              <goal>testCompile</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-assembly-plugin</artifactId>
+        <version>3.0.0</version>
+        <configuration>
+          <descriptorRefs>
+            <descriptorRef>jar-with-dependencies</descriptorRef>
+          </descriptorRefs>
+        </configuration>
+        <executions>
+          <execution>
+            <id>make-assembly</id>
+            <phase>package</phase>
+            <goals>
+              <goal>single</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>

src/main/resources/hello.txt

@@ -0,0 +1,6 @@
+hello flink
+hello word
+hello scala
+how are you
+fine thank you
+and you

src/main/scala/com/zhangblue/wc/StreamWordCount.scala

@@ -0,0 +1,38 @@
+package com.zhangblue.wc
+
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.streaming.api.scala._
+
+/**
+ * 流式处理word count
+ */
+object StreamWordCount {
+
+  def main(args: Array[String]): Unit = {
+
+    //通过参数进行传递. 参数传递方式：--host localhost --port 7777
+    val parameter = ParameterTool.fromArgs(args)
+    val host:String  = parameter.get("host")
+    val port:Int = parameter.getInt("port")
+
+
+    //创建一个流处理的执行环境
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+
+    //接收socket数据流
+    val testDataStream = env.socketTextStream(host,port)
+
+    //逐一读取数据
+   val wordCountDataStream = testDataStream.flatMap(_.split(" "))
+     .filter(_.nonEmpty)
+     .map((_,1))
+     .keyBy(0)
+     .sum(1)
+
+    //打印输出
+    wordCountDataStream.print()
+
+    //执行任务
+    env.execute("Stream WordCount Example")
+  }
+}

src/main/scala/com/zhangblue/wc/WordCount.scala

@@ -0,0 +1,27 @@
+package com.zhangblue.wc
+
+import org.apache.flink.api.scala._
+
+//批处理word count
+object WordCount {
+
+  def main(args: Array[String]): Unit = {
+    //创建一个批处理的执行环境
+    val env = ExecutionEnvironment.getExecutionEnvironment
+   //从文件中读取数据
+    val inputPath = "D:\\software\\workspace\\FlinkTutorial\\src\\main\\resources\\hello.txt"
+    val inputDataSet = env.readTextFile(inputPath)
+
+    //分词之后做count
+   val wordCountDataSet = inputDataSet.flatMap(_.split(" "))
+     .map((_,1))
+     .groupBy(0)
+     .sum(1)
+
+
+    //打印输出
+    wordCountDataSet.print()
+  }
+
+
+}