Flink实战（二）开发一个简单的程序 WordCount

约定不等于承诺〃 2023-10-11 10:00 26阅读 0赞

### 文章目录 ###

*   *  开发
     *   *  搭建开发环境
         *  启动应用程序
         *   *  1. 启动nc
             *  2. 启动程序

## 开发 ##

### 搭建开发环境 ###

1.  准备 `pom.xml` 文件

<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.wuhulala.flink</groupId>
    	<artifactId>chapter1</artifactId>
    	<version>1.0-SNAPSHOT</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.9.0</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>org.slf4j</groupId>
    			<artifactId>slf4j-log4j12</artifactId>
    			<version>1.7.7</version>
    			<scope>runtime</scope>
    		</dependency>
    		<dependency>
    			<groupId>log4j</groupId>
    			<artifactId>log4j</artifactId>
    			<version>1.2.17</version>
    			<scope>runtime</scope>
    		</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>com.wuhulala.flink.StreamingJob</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>

### 启动应用程序 ###

#### 1. 启动nc ####

在windows上启动命令和linux稍许不一样,多加一个`-p`参数

nc -L -p 18080

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

#### 2. 启动程序 ####

package com.wuhulala.flink;
    
    import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.common.functions.ReduceFunction;
    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 { 
    
            // the port to connect to
            final int port = 18080;
    // try { 
    // final ParameterTool params = ParameterTool.fromArgs(args);
    // port = params.getInt("port");
    // } catch (Exception e) { 
    // System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'");
    // return;
    // }
    
            // get the execution environment
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 连接到对应nc 开启的端口上
            DataStream<String> text = env.socketTextStream("localhost", port, "\n");
    
            // parse the data, group it, window it, and aggregate the counts
            DataStream<WordWithCount> windowCounts = text
                .flatMap(new FlatMapFunction<String, WordWithCount>() { 
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) { 
                        for (String word : value.split("\\s")) { 
                            out.collect(new WordWithCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                // 事件窗口 窗口大小，窗口刻度
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .reduce(new ReduceFunction<WordWithCount>() { 
                    @Override
                    public WordWithCount reduce(WordWithCount a, WordWithCount b) { 
                        return new WordWithCount(a.word, a.count + b.count);
                    }
                });
    
            // print the results with a single thread, rather than in parallel
            windowCounts.print().setParallelism(1);
    
            env.execute("Socket Window WordCount");
        }
    
        // Data type for words with count
        public static class WordWithCount { 
    
            public String word;
            public long count;
    
            public WordWithCount() { }
    
            public WordWithCount(String word, long count) { 
                this.word = word;
                this.count = count;
            }
    
            @Override
            public String toString() { 
                return word + " : " + count;
            }
        }
    }

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

就这样，一个简单的flink程序就完成了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190922164230794.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190922164301759.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ=,size_16,color_FFFFFF,t_70