路人王
一、环境准备
1.下载地址：https://flink.apache.org
下面那个hadoop的整合包要放到flink的lib中去。
2.上传到linux中去，并解压到相关目录
tar -zxvf flink-1.9.1-... apps/
二、standalone部署
2.1 修改conf中的flink-conf.yaml
# 1.主节点的主机名
jobmanager.rpc.address: hadoop01
# 2.节点的资源槽数
taskmanager.numberOfTaskSlots: 2
# 3.单机的话，暂时不用配置zookeeper的地址
2.2 修改conf中的slaves
# 设置从节点
hadoop02
hadoop03
2.3 拷贝到其他节点
scp -r flink-1.9.1/ hadoop02:$PWD
scp -r flink-1.9.1/ hadoop03:$PWD
2.4 启动集群
bin/start-cluster.sh
2.5 测试访问
hadoop01:8081
2.6 页面提交程序jar包
提前开设端口，在上面的提交之前。
2.7 命令行窗口提交程序jar包
# --hostname hadoop01 --port 8888 为参数
bin/flink run -m hadoop01 -p 4 -c com.wang.Main.class /jar路径 --hostname hadoop01 --port 8888
三、项目整合【Maven3.x+Jdk8/Scala-2.11】
3.1 pom依赖
<dependencies>
<!--如果是java程序 java所需要的jar-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.9.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.9.1</version>
<scope>provided</scope>
</dependency>
<!--如果是scala程序 scala所需要的jar-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.9.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.9.1</version>
<scope>provided</scope>
</dependency>
<!--依赖日志-->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- 编译插件 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.6.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<!-- scala编译插件 -->
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.1.6</version>
<configuration>
<scalaCompatVersion>2.11</scalaCompatVersion>
<scalaVersion>2.11.12</scalaVersion>
<encoding>UTF-8</encoding>
</configuration>
<executions>
<execution>
<id>compile-scala</id>
<phase>compile</phase>
<goals>
<goal>add-source</goal>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>test-compile-scala</id>
<phase>test-compile</phase>
<goals>
<goal>add-source</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- 打jar包插件(会包含所有依赖) -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.6</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<!-- 可以设置jar包的入口类(可选) -->
<mainClass>com.wang.flink.SocketWordCount</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
3.2 实时计算【DataStream】
1.利用socket通信实现实时的单词计数计算。 2.开启socket端口号
nc -lk 8888
3.java程序
public class StreamingWordCount {
public static void main(String[] args) throws Exception {
// 1.创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2.创建数据集Source
DataStream<String> lines = env.socketTextStream("192.168.52.200", 8888);
// 3.数据转换Transformations
DataStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String line, Collector<String> out) throws Exception {
String[] words = line.split(" ");
for (String word : words) {
// 输出到收集器
out.collect(word);
}
}
});
// 4.单词和1组合
SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String word) throws Exception {
return Tuple2.of(word, 1);
}
});
// 5. 分组聚合 单词：次数
SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = wordAndOne.keyBy(0).sum(1);
// 6.Sink 数据下沉
// 这里只打印到控制台
sumed.print();
// 7.启动
env.execute("StreamingWordCount");
}
}
3.测试结果
4.打包到集群运行
将程序中的socket通信的主机地址和端口号改为参数形式如：
DataStream<String> lines = env.socketTextStream(args[0], Integer.parseInt(args[1]));
1)web页面提交,提前开启socket端口
测试结果如下：
5.java8 lambda表达式的优化
SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
Arrays.stream(line.split(" ")).forEach(w -> {
out.collect(Tuple2.of(w, 1));
});
});
3.3 离线计算【DataSet】
1.整理一个文件，放入一些数据如
flink flink spark hadoop
flink vue java
hdfs spark
2.java程序
public class BatchWordCount {
public static void main(String[] args) throws Exception {
// 1.获取配置
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 2.读取数据
DataSource<String> lines = env.readTextFile(args[0]);
// 3.切分压平
FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] words = line.split(" ");
for (String word : words) {
out.collect(Tuple2.of(word, 1));
}
}
});
// 4.离线计算 分组聚合groupBy()，而不是实时计算的keyBy()
AggregateOperator<Tuple2<String, Integer>> sumed = wordAndOne.groupBy(0).sum(1);
// 5.保存数据 设置并行度 数据几个文件
sumed.writeAsText(args[1]).setParallelism(2);
// 6.执行
env.execute("BatchWordCount");
}
}
评论