docker flink集群中运行socketWordcount项目

我不是女神ヾ 2023-01-04 11:53 157阅读 0赞

## 使用mvn构建项目 ##

使用mvn命令，输入如下命令

mvn archetype:generate \
        -DarchetypeGroupId=org.apache.flink \
        -DarchetypeArtifactId=flink-quickstart-java \
        -DarchetypeVersion=1.6.1 \
        -DgroupId=my-flink-project \
        -DartifactId=my-flink-project \
        -Dversion=0.1 \
        -Dpackage=myflink \
        -DinteractiveMode=false

等待构建完毕后，如下图所示

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center]

如果速度太慢，需要到maven的安装目录==>conf==>setting.xml里添加阿里的镜像站。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 1]

## 使用IDEA编写flink程序 ##

在idea中导入刚刚创建的maven项目，修改pom文件如下

<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>my-flink-project</groupId>
    	<artifactId>my-flink-project</artifactId>
    	<version>0.1</version>
    	<packaging>jar</packaging>
    
    	<name>Flink Quickstart Job</name>
    	<url>http://www.myorganization.org</url>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<flink.version>1.11.2</flink.version>
    		<java.version>1.8</java.version>
    		<scala.binary.version>2.11</scala.binary.version>
    		<maven.compiler.source>${java.version}</maven.compiler.source>
    		<maven.compiler.target>${java.version}</maven.compiler.target>
    	</properties>
    
    	<repositories>
    		<repository>
    			<id>apache.snapshots</id>
    			<name>Apache Development Snapshot Repository</name>
    			<url>https://repository.apache.org/content/repositories/snapshots/</url>
    			<releases>
    				<enabled>false</enabled>
    			</releases>
    			<snapshots>
    				<enabled>true</enabled>
    			</snapshots>
    		</repository>
    	</repositories>
    
    	<dependencies>
    		
    		
    		<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-java</artifactId>
    			<version>${flink.version}</version>
    			<scope>provided</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
    			<version>${flink.version}</version>
    			<scope>provided</scope>
    		</dependency>
    
    		
    
    		
    
    		
    		
    
    		<dependency>
    			<groupId>log4j</groupId>
    			<artifactId>log4j</artifactId>
    			<version>1.2.17</version>
    			<scope>runtime</scope>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.apache.maven.plugins</groupId>
    			<artifactId>maven-shade-plugin</artifactId>
    			<version>3.0.0</version>
    		</dependency>
    
    	</dependencies>
    
    	<build>
    		<plugins>
    
    			
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-compiler-plugin</artifactId>
    				<version>3.1</version>
    				<configuration>
    					<source>${java.version}</source>
    					<target>${java.version}</target>
    				</configuration>
    			</plugin>
    
    			
    			
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-shade-plugin</artifactId>
    				<version>3.0.0</version>
    				<executions>
    					
    					<execution>
    						<phase>package</phase>
    						<goals>
    							<goal>shade</goal>
    						</goals>
    						<configuration>
    							<artifactSet>
    								<excludes>
    									<exclude>org.apache.flink:force-shading</exclude>
    									<exclude>com.google.code.findbugs:jsr305</exclude>
    									<exclude>org.slf4j:*</exclude>
    									<exclude>log4j:*</exclude>
    								</excludes>
    							</artifactSet>
    							<filters>
    								<filter>
    									
    									<artifact>*:*</artifact>
    									<excludes>
    										<exclude>META-INF/*.SF</exclude>
    										<exclude>META-INF/*.DSA</exclude>
    										<exclude>META-INF/*.RSA</exclude>
    									</excludes>
    								</filter>
    							</filters>
    							<transformers>
    								<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
    									<mainClass>myflink.SocketWindowWordCount</mainClass>
    								</transformer>
    							</transformers>
    						</configuration>
    					</execution>
    				</executions>
    			</plugin>
    		</plugins>
    
    		<pluginManagement>
    			<plugins>
    
    				
    				<plugin>
    					<groupId>org.eclipse.m2e</groupId>
    					<artifactId>lifecycle-mapping</artifactId>
    					<version>1.0.0</version>
    					<configuration>
    						<lifecycleMappingMetadata>
    							<pluginExecutions>
    								<pluginExecution>
    									<pluginExecutionFilter>
    										<groupId>org.apache.maven.plugins</groupId>
    										<artifactId>maven-shade-plugin</artifactId>
    										<versionRange>[3.0.0,)</versionRange>
    										<goals>
    											<goal>shade</goal>
    										</goals>
    									</pluginExecutionFilter>
    									<action>
    										<ignore/>
    									</action>
    								</pluginExecution>
    								<pluginExecution>
    									<pluginExecutionFilter>
    										<groupId>org.apache.maven.plugins</groupId>
    										<artifactId>maven-compiler-plugin</artifactId>
    										<versionRange>[3.1,)</versionRange>
    										<goals>
    											<goal>testCompile</goal>
    											<goal>compile</goal>
    										</goals>
    									</pluginExecutionFilter>
    									<action>
    										<ignore/>
    									</action>
    								</pluginExecution>
    							</pluginExecutions>
    						</lifecycleMappingMetadata>
    					</configuration>
    				</plugin>
    			</plugins>
    		</pluginManagement>
    	</build>
    
    	
    	
    	
    	<profiles>
    		<profile>
    			<id>add-dependencies-for-IDEA</id>
    
    			<activation>
    				<property>
    					<name>idea.version</name>
    				</property>
    			</activation>
    
    			<dependencies>
    				<dependency>
    					<groupId>org.apache.flink</groupId>
    					<artifactId>flink-java</artifactId>
    					<version>${flink.version}</version>
    					<scope>compile</scope>
    				</dependency>
    				<dependency>
    					<groupId>org.apache.flink</groupId>
    					<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
    					<version>${flink.version}</version>
    					<scope>compile</scope>
    				</dependency>
    			</dependencies>
    		</profile>
    	</profiles>
    
    </project>

在myflink包下创建java类SocketWindowWordCount，代码如下

/*
     * Licensed to the Apache Software Foundation (ASF) under one
     * or more contributor license agreements.  See the NOTICE file
     * distributed with this work for additional information
     * regarding copyright ownership.  The ASF licenses this file
     * to you under the Apache License, Version 2.0 (the
     * "License"); you may not use this file except in compliance
     * with the License.  You may obtain a copy of the License at
     *
     *     http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package myflink;
    
    import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    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 {
        
    
            // 创建 execution environment
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 通过连接 socket 获取输入数据，这里连接到本地9000端口，如果9000端口已被占用，请换一个端口
            DataStream<String> text = env.socketTextStream("master", 9000, "\n");
    
            // 解析数据，按 word 分组，开窗，聚合
            DataStream<Tuple2<String, Integer>> windowCounts = text
                    .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
        
                        @Override
                        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        
                            for (String word : value.split("\\s")) {
        
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                    })
                    .keyBy(0)
                    .timeWindow(Time.seconds(5))
                    .sum(1);
    
            // 将结果打印到控制台，注意这里使用的是单线程打印，而非多线程
            windowCounts.print().setParallelism(1);
    //        System.out.println(windowCounts);
    
            env.execute("Socket Window WordCount");
        }
    }

然后整个项目打包，需要注意的是，这里的diretory for META-INF/MANIFEST.MF默认到java文件夹，我们修改到src，如下图所示

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 2]

这样，整个项目的结构如下图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 3]

如果没有修改这里的话，运行的时候就会报错

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 4]

## 集群中运行测试 ##

使用docker cp命令将本地的jar包传到master节点中

安装nc

yum install nc -y

启动集群

start-cluster.sh

先启动netcat（注意要先启动这个，不能先运行jar包）

nc -lk 9000

然后再打开一个终端，运行jar包

flink run /myjars/my-flink-project.jar

出现

![在这里插入图片描述][20210109185721838.png_pic_center]

就是说明程序正在运行了

webUI中也可以看到程序正在运行

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 5]

在第一个终端中持续输入单词并回车，终端中是没有打印输出的，可以在webUI的taskmanager的sdout中查看输出，或者在loglist中的\*.out中查看。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 6]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 7]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 8]

注意我们输出是一个线程，所以只在一个taskmanager有输出。

在第一个终端（netcat终端）Ctrl+c就可以停止程序了。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 9]

## IDEA中运行测试 ##

修改pom文件

添加一个依赖

<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-clients_${scala.binary.version}</artifactId>
    			<version>${flink.version}</version>
    		</dependency>

修改下图将provided改为compile（否则会报JNI错误）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 10]

安装netcat

访问https://eternallybored.org/misc/netcat/

点击下图的位置下载

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 11]

解压后将nc.exe放到window的system32目录下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 12]

在cmd中输入

nc -l -p 9000

idea中运行socketwordcount

然后在cmd中输入单词

结果如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 13]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221119/9ba20db037be4a8daff563e61c925d20.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221119/02d669a956dd48ecbf72d2fde0b80c45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20210109185531738.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20210109185602221.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20210109185647817.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[20210109185721838.png_pic_center]: https://img-blog.csdnimg.cn/20210109185721838.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20210109185737280.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20210109185811447.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20210109185828235.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 8]: https://img-blog.csdnimg.cn/20210109185856351.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 9]: https://img-blog.csdnimg.cn/202101091859362.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 10]: https://img-blog.csdnimg.cn/20210109222951410.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 11]: https://img-blog.csdnimg.cn/20210109223014573.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 12]: https://img-blog.csdnimg.cn/20210109223119970.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1_size_16_color_FFFFFF_t_70_pic_center 13]: https://img-blog.csdnimg.cn/20210109223138601.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM0NTA1,size_16,color_FFFFFF,t_70#pic_center