编写最简单flink应用并提交到flink v1.19.2集群

1 概述

本文介绍编写最最简单的word count的代码，编译成jar后，提交到flink v1.19.2集群进行运行。

2 环境准备

2.1 jdk和maven工具的安装

yum安装jdk 1.8：

yum install java-1.8.0-openjdk-1.8.0.212.b04-0.el7_6.x86_64 java-1.8.0-openjdk-devel-1.8.0.212.b04-0.el7_6.x86_64 -y

下载maven二进制文件的压缩包，并解压。

cd /usr/local
wget https://dlcdn.apache.org/maven/maven-3/3.9.9/binaries/apache-maven-3.6.3-bin.tar.gz
tar xf apache-maven-3.6.3-bin.tar.gz
ln -s /usr/local/apache-maven-3.6.3 apache-maven

将以下两行追加到/etc/profile：

MAVEN_HOME=/usr/local/apache-maven
PATH=$PATH:$MAVEN_HOME/bin

加载linux环境变量：

source /etc/profile

查看maven的信息：
在这里插入图片描述

flink_29">2.2 准备flink集群

https://archive.apache.org/dist/flink/flink-1.19.2/

下载flink安装包，并启动集群，启动脚本放在bin目录中，脚本名称为start-cluster.sh。
在这里插入图片描述
flink集群的web控制台监听在8081端口：

flink_39">3 编写flink应用代码

3.1 创建代码目录

cd /tmp
mkdir -p wordcount
cd wordcount
mkdir -p src/main/java/org/example

3.2 pom.xml文件

pom.xml内容如下，可直接拿来使用。
pom.xml指定flink版本为1.19.2，只需显式指明两个依赖flink-streaming-java和flink-clients。

<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>org.example</groupId>
    <artifactId>flink-demo</artifactId>
    <version>1.0-SNAPSHOT</version>


    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.19.2</flink.version>
    </properties>


    <dependencies>
        <!-- Flink Streaming Java API -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>


        <!-- Flink CLI Support (optional, for local execution) -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>org.example.WordCount</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

3.3 java代码

文件名称为src/main/java/org/example/WordCount.java，内容如下：

package org.example;

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.util.Collector;

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 创建流处理执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // 定义数据源（从集合中读取数据）
        DataStream<String> text = env.fromElements(
                "Hello Flink",
                "Hello World",
                "Flink is awesome",
                "World is big"
        );


        // 转换操作：将句子拆分为单词，并统计每个单词的出现次数
        DataStream<Tuple2<String, Integer>> counts = text
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0) // 按单词分组
                .sum(1); // 对单词计数求和


        // 输出结果
        counts.print();


        // 启动任务
        env.execute("WordCount Example");
    }


    // 自定义 FlatMapFunction，将句子拆分为单词
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 将句子转换为小写并按空格分割
            String[] words = value.toLowerCase().split("\\W+");


            // 发射每个单词及其初始计数（1）
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}