Flink1.12.0简单实现wordcount

àì夳堔傛蜴生んèń 2022-11-15 11:22 223阅读 0赞

# 文章目录 #

前言

一、Flink1.12.0简单实现wordcount

二、使用步骤

1.引入pom.xml

2.主类

3.运行结果

总结

--------------------

# 前言 #

Flink1.12.0简单实现wordcount

--------------------

# 一、Flink1.12.0简单实现wordcount #

Flink1.12.0简单实现wordcount

# 二、使用步骤 #

## 1.引入pom.xml ##

代码如下（示例）：

<?xml version="1.0" encoding="UTF-8"?>
    <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">
        <parent>
            <artifactId>flinkdemo</artifactId>
            <groupId>com.flink.pangsl</groupId>
            <version>1.0-SNAPSHOT</version>
    
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>frauddetection</artifactId>
        <packaging>jar</packaging>
    
        <name>Flink Walkthrough DataStram Scala</name>
        <url>https://flink.apache.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.12.0</flink.version>
            <scala.binary.version>2.11</scala.binary.version>
            <scala.version>2.11.12</scala.version>
            <log4j.version>2.12.1</log4j.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </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.apache.flink</groupId>
                <artifactId>flink-clients_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>compile</scope>
            </dependency>
    
            
    
            
    
            
            
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-slf4j-impl</artifactId>
                <version>${log4j.version}</version>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-api</artifactId>
                <version>${log4j.version}</version>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-core</artifactId>
                <version>${log4j.version}</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>org.apache.logging.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>spendreport.FraudDetectionJob</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>
                    <configuration>
                        <args>
                            <arg>-nobootcp</arg>
                        </args>
                    </configuration>
                </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.9.1</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>
    
    </project>

## 2.主类 ##

代码如下（示例）：

package wordcount
    
    import org.apache.flink.api.scala.ExecutionEnvironment
    
    /**
     * @author pangsl
     * @date 2021/3/31 15:38
     */
    object WordCount {
      def main(args: Array[String]): Unit = {
    
          //1.准备环境-env,注意导scala包
          val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    
          //2.数据源-Source
          import org.apache.flink.api.scala._
          val linesDS: DataSet[String] = env.fromElements("Flink Spark Hadoop", "Flink Spark", "Flink")
    
          //3.数据转换-Transformation   每一行按照空格切分并压平
          val wordDS: DataSet[String] = linesDS.flatMap(_.split(" "))
          //每个单词记为1
          val wordAndOneDS: DataSet[(String, Int)] = wordDS.map((_, 1))
          //分组
          val groupedDS: GroupedDataSet[(String, Int)] = wordAndOneDS.groupBy(0) //0表示按照索引为0的元素即单词进行分组
          //注意:聚合不支持key选择器函数,也就是不支持_._1,否则报错:Aggregate does not support grouping with KeySelector functions, yet.
          //val groupedDS: GroupedDataSet[(String, Int)] = wordAndOneDS.groupBy(_._1)
          //聚合
          val result: AggregateDataSet[(String, Int)] = groupedDS.sum(1) //1表示按照索引为1的元素即数量进行聚合
    
          //4.数据输出-Sink
          result.print() //print可以触发执行
    
          //5.调用执行操作
          //env.execute()//latest call to 'execute()', 'count()', 'collect()', or 'print()'
    
        }
    
    }

## 3.运行结果 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

--------------------

# 总结 #

Flink1.12.0简单实现wordcount，并打印输出。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221022/0d0d18082c244ef484ffa41ecfa2b92e.png