目录

1.Flink项目初始化

2.Hello-world

---

1.Flink项目初始化

新建maven项目或者gradle项目，这里使用maven项目为例。

在项目的pom.xml文件中添加Flink依赖，如下所示，为Hello-World例子的最小依赖：

 <properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><!-- 指定Flink版本 --><flink.version>1.12.2</flink.version> <!-- 请根据需要调整版本号 --><scala.binary.version>2.12</scala.binary.version> <!-- 如果使用Scala，请指定Scala版本 --><java.version>1.8</java.version> <!-- 或者更高的Java版本 --></properties><dependencies><!-- Apache Flink dependencies --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!-- 如果你打算使用Table API或SQL，还需要添加以下依赖 --><!-- 其他可能需要的依赖，如连接器、格式等 --><!-- 测试依赖 --><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.13.2</version><scope>test</scope></dependency></dependencies>

为了加速依赖的下载，尤其是在国内，可以配置maven的国内镜像，配置方法为设置maven的settings.xml文件。如果是mac用户，该文件位于家目录下.m2/settings.xml。不存在可以vim一个，早文件中添加以下内容。

<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd"><mirrors><!-- 配置阿里云Maven镜像 --><mirror><id>aliyunmaven</id><mirrorOf>central</mirrorOf><name>Aliyun Maven</name><url>https://maven.aliyun.com/repository/public</url></mirror><!-- 如果还需要其他镜像，可以继续添加 --></mirrors>
</settings>

2.Hello-world

还是一如既往的大数据经典案例：Hello World

代码如下所示：

package com.wlh.p1;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class HelloWorld {public static void main(String[] args) throws Exception {//StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.socketTextStream("localhost", 7777).flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {String[] split = s.split(" ");for (String word : split) {collector.collect(Tuple2.of(word, 1));}}}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {return stringIntegerTuple2.f0;}}).sum(1).print();//env.execute();}
}

代码不是很复杂，Flink的api相对而言和Spark比较类似，先创建一个上下文的执行环境，在项目结束的位置，使用execute运行，Spark中也有collect操作。上面的代码就是一个计算Word-Count的例子。

当我们输入：

hello world
hello rust
hello spark

程序输出：

4> (hello,1)
6> (world,1)
7> (rust,1)
4> (hello,2)
4> (hello,3)
1> (spark,1)

对于输出的一些解释，对于输出我们做一些解释，可以看出流计算和批计算的差异

输入hello world后，输出(hello,1) (world,1)

输入hello rust后，输出(hello,2) (rust,1)

输入hello spark后，输出(hello,3) (spark,1)

通过上面的计算输出，可以看出流计算的特点

a.立即计算，收到一条，计算一次

b.有状态，通过对hello的追踪，可以看出计算时是可以读写状态的