Flink入门(一)(Java和scala)

「爱情、让人受尽委屈。」 2024-04-18 17:11 51阅读 0赞

## Flink批处理 ##

从文件中读取单词，计算频次

### Scala版本 ###

import org.apache.flink.api.scala.ExecutionEnvironment
    
    object BatchWordCountJob {
    
      def main(args: Array[String]) {
        val env = ExecutionEnvironment.getExecutionEnvironment
        val input ="/Users/mac126/Desktop/hello.txt";
    
        val  text = env.readTextFile(input)
        import org.apache.flink.api.scala._
        text.flatMap(_.toLowerCase().split("\t")).filter(_.nonEmpty).map((_,1))
          .groupBy(0).sum(1).print()
      }

最后可在控制台看到打印的内容

### Java版本 ###

mport org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.util.Collector;
    
    /**
     * java开发wordcount
     */
    public class BatchWordCountJob {
    
    
    
        public static void main(String[] args) throws Exception{
    
    
            final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            String file="/Users/mac126/Desktop/hello.txt";
    
            DataSet<String>  text = env.readTextFile(file);
    
            text.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String,Integer>> collector) throws Exception {
                        String [] tokens = value.toLowerCase().split("\t");
                        for (String token :tokens){
                            if (token.trim().length()>0){
                                collector.collect(new Tuple2<String, Integer>(token,1));
                            }
                        }
                }
            }).groupBy(0).sum(1).print();
    
        }
    }

## Flink流处理 ##

flink入门一般使用socket数据作为流式数据来源。在mac上命令行使用nc -lk 9000开启socket发送端，在flink中接收socket消息如下

### Scala版本 ###

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.windowing.time.Time
    
    object StreamWordCountJob {
      def main(args: Array[String]) {
        // set up the streaming execution environment
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val  text =env.socketTextStream("localhost",9000)
        import org.apache.flink.api.scala._
        text.flatMap(_.split(",")).map((_,1)).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print().setParallelism(1)
        env.execute("StreamWordCountJob")
      }
    }

### Java版本 ###

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    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 StreamWordCountJob {
        public static void main(String[] args) throws Exception{
    
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStreamSource<String> text = env.socketTextStream("localhost",9000);
            text.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                    String [] tokens = value.toLowerCase().split(",");
                    for (String token:tokens){
                        if (token.length()>0){
                            collector.collect(new Tuple2<String, Integer>(token,1));
                        }
                    }
                }
            }).keyBy(0).sum(1).print().setParallelism(1);
            env.execute("StreamWordCountJob");
    
        }
    }

以上就是Flink入门的第一个小例子。大家可以在自己电脑上试下。

## 前置pom文件 ##

一般在IDEA上安装Scala插件进行开发

### scala版本 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.quickstart</groupId>
    	<artifactId>quickstart</artifactId>
    	<version>0.1</version>
    	<packaging>jar</packaging>
    
    	<name>Flink Quickstart Job</name>
    	<url>http://www.myorganization.org</url>
    
    	<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>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<flink.version>1.8.0</flink.version>
    		<scala.binary.version>2.11</scala.binary.version>
    		<scala.version>2.11.12</scala.version>
    	</properties>
    
    	<dependencies>
    		
    		
    		<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-scala_${scala.binary.version}</artifactId>
    			<version>${flink.version}</version>
    			<scope>provided</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.flink</groupId>
    			<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
    			<version>${flink.version}</version>
    			<scope>provided</scope>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.scala-lang</groupId>
    			<artifactId>scala-library</artifactId>
    			<version>${scala.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-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>my.flink.quickstart.StreamingJob</mainClass>
    								</transformer>
    							</transformers>
    						</configuration>
    					</execution>
    				</executions>
    			</plugin>
    
    			
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-compiler-plugin</artifactId>
    				<version>3.1</version>
    				<configuration>
    					<source>1.8</source>
    					<target>1.8</target>
    				</configuration>
    			</plugin>
    
    			
    			<plugin>
    				<groupId>net.alchim31.maven</groupId>
    				<artifactId>scala-maven-plugin</artifactId>
    				<version>3.2.2</version>
    				<executions>
    					<execution>
    						<goals>
    							<goal>compile</goal>
    							<goal>testCompile</goal>
    						</goals>
    					</execution>
    				</executions>
    			</plugin>
    
    			
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-eclipse-plugin</artifactId>
    				<version>2.8</version>
    				<configuration>
    					<downloadSources>true</downloadSources>
    					<projectnatures>
    						<projectnature>org.scala-ide.sdt.core.scalanature</projectnature>
    						<projectnature>org.eclipse.jdt.core.javanature</projectnature>
    					</projectnatures>
    					<buildcommands>
    						<buildcommand>org.scala-ide.sdt.core.scalabuilder</buildcommand>
    					</buildcommands>
    					<classpathContainers>
    						<classpathContainer>org.scala-ide.sdt.launching.SCALA_CONTAINER</classpathContainer>
    						<classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
    					</classpathContainers>
    					<excludes>
    						<exclude>org.scala-lang:scala-library</exclude>
    						<exclude>org.scala-lang:scala-compiler</exclude>
    					</excludes>
    					<sourceIncludes>
    						<sourceInclude>**/*.scala</sourceInclude>
    						<sourceInclude>**/*.java</sourceInclude>
    					</sourceIncludes>
    				</configuration>
    			</plugin>
    			<plugin>
    				<groupId>org.codehaus.mojo</groupId>
    				<artifactId>build-helper-maven-plugin</artifactId>
    				<version>1.7</version>
    				<executions>
    					
    					<execution>
    						<id>add-source</id>
    						<phase>generate-sources</phase>
    						<goals>
    							<goal>add-source</goal>
    						</goals>
    						<configuration>
    							<sources>
    								<source>src/main/scala</source>
    							</sources>
    						</configuration>
    					</execution>
    					
    					<execution>
    						<id>add-test-source</id>
    						<phase>generate-test-sources</phase>
    						<goals>
    							<goal>add-test-source</goal>
    						</goals>
    						<configuration>
    							<sources>
    								<source>src/test/scala</source>
    							</sources>
    						</configuration>
    					</execution>
    				</executions>
    			</plugin>
    		</plugins>
    	</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-scala_${scala.binary.version}</artifactId>
    					<version>${flink.version}</version>
    					<scope>compile</scope>
    				</dependency>
    				<dependency>
    					<groupId>org.apache.flink</groupId>
    					<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
    					<version>${flink.version}</version>
    					<scope>compile</scope>
    				</dependency>
    				<dependency>
    					<groupId>org.scala-lang</groupId>
    					<artifactId>scala-library</artifactId>
    					<version>${scala.version}</version>
    					<scope>compile</scope>
    				</dependency>
    			</dependencies>
    		</profile>
    	</profiles>
    
    </project>

### Java版本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.6.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>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>myflink.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>