辑环境变量：

sudo vim ~/.bashrc
增加Flink的路径

2、使环境变量生效：

#使环境变量生效
source ~/.bashrc
#查看环境变量
echo $Flink_HOME
修改Kafka、Flink以及DolphinScheduler的配置文件
因为用的是虚拟机，为了让外面的主机能够访问到虚拟机的网络，需要修改下配置文件

修改Kafka配置：找到Kafka安装包下的config文件夹，修改config下的server.properties文件，修改listeners是为了外面的主机能够访问到虚拟机的Kafka，还有把advertised.listeners改成虚拟机地址，写样例的时候能连上虚拟机的Kafka地址，不然默认连localhost
broker.id=0
listeners=PLAINTEXT://0.0.0.0:9092
#192.168.146.132修改成虚拟机ip
advertised.listeners=PLAINTEXT://192.168.146.132:9092

修改Flink配置：找到Flink安装包下的conf文件夹，修改conf下的Flink-conf.yaml文件，把里面所有的localhost地址全部改成0.0.0.0，以便主机能访问到虚拟机的Flink。还有增加jobmanager和taskmanager的内存
jobmanager.rpc.address: 0.0.0.0
jobmanager.bind-host: 0.0.0.0
jobmanager.cpu.cores: 1
jobmanager.memory.process.size: 1600m
taskmanager.bind-host: 0.0.0.0
taskmanager.host: 0.0.0.0
taskmanager.memory.process.size: 2048m
taskmanager.cpu.cores: 1

修改Apache DolphinScheduler的配置文件，从Apache DolphinScheduler的启动脚本文件dolphinscheduler-daemon.sh可以看出，配置环境变量用的是bin/env文件夹下的dolphinscheduler_env.sh。
查看dolphinscheduler-daemon.sh文件：

修改dolphinscheduler_env.sh文件，新增JAVA、Flink路径：

#修改成自己的JAVA、Flink路径
export JAVA_HOME=/data/jdk-11.0.29
export Flink_HOME=/data/Flink-1.18.1

关闭防火墙，启动应用
启动应用，包括Zookeeper、Kafka、Flink以及Apache DolphinScheduler。

#关闭防火墙
sudo systemctl stop firewalld

在 Flink 根目录下，执行以下命令启动 Flink 集群

bin/start-cluster.sh

启动 ZooKeeper

bin/zookeeper-server-start.sh config/zookeeper.properties &

启动 Kafka 服务器

bin/Kafka-server-start.sh config/server.properties &

#创建 Kafka 主题
bin/Kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

#使用命令行生产者发送消息
bin/Kafka-console-producer.sh --topic test --bootstrap-server localhost:9092

#消费
bin/Kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092

启动 Standalone Server 服务

bash ./bin/dolphinscheduler-daemon.sh start standalone-server
测试
测试Flink、Apache DolphinScheduler是否能访问成功。

Flink访问地址： http://localhost:8081/ ，localhost改成自己虚拟机地址

Apache DolphinScheduler访问地址：http://localhost:12345/dolphinscheduler/ui ，localhost改成自己虚拟机地址即可登录系统 UI。默认的用户名和密码是 admin/dolphinscheduler123

编写样例
用Flink消费Kafka数据，然后打包上传到Apache DolphinScheduler，启动Flink任务：

编写样例：
pom.xml


4.0.0

<groupId>com.example</groupId>
<artifactId>Flink-Kafka-demo</artifactId>
<version>1.0-SNAPSHOT</version>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <Flink.version>1.18.1</Flink.version>
    <scala.binary.version>2.12</scala.binary.version>
    <Kafka.version>3.6.0</Kafka.version>
</properties>

<dependencies>
    <!-- Flink核心依赖 -->
    <dependency>
        <groupId>org.apache.Flink</groupId>
        <artifactId>Flink-java</artifactId>
        <version>${Flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.Flink</groupId>
        <artifactId>Flink-streaming-java</artifactId>
        <version>${Flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.Flink</groupId>
        <artifactId>Flink-clients</artifactId>
        <version>${Flink.version}</version>
    </dependency>

    <!-- 连接器基础依赖 -->
    <dependency>
        <groupId>org.apache.Flink</groupId>
        <artifactId>Flink-connector-base</artifactId>
        <version>${Flink.version}</version>
    </dependency>

    <!-- Kafka连接器（关键修改点） -->
    <dependency>
        <groupId>org.apache.Flink</groupId>
        <artifactId>Flink-connector-Kafka</artifactId>
        <version>3.1.0-1.18</version>
    </dependency>
    <dependency>
        <groupId>org.apache.Kafka</groupId>
        <artifactId>Kafka-clients</artifactId>
        <version>${Kafka.version}</version>
    </dependency>

    <!-- 日志依赖 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.36</version>
        <scope>runtime</scope>
    </dependency>
</dependencies>

<repositories>
    <repository>
        <id>aliyun</id>
        <url>https://maven.aliyun.com/repository/public</url>
        <releases>
            <enabled>true</enabled>
        </releases>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>apache-releases</id>
        <url>https://repository.apache.org/content/repositories/releases/</url>
    </repository>
</repositories>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>${maven.compiler.source}</source>
                <target>${maven.compiler.target}</target>
            </configuration>
        </plugin>
        <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>
                        <artifactSet>
                            <excludes>
                                <exclude>org.apache.Flink:force-shading</exclude>
                                <exclude>com.google.code.findbugs:jsr305</exclude>
                                <exclude>org.slf4j:*</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>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>
FlinkKafkaConsumerExample.java

import org.apache.Flink.api.common.functions.FlatMapFunction;
import org.apache.Flink.api.java.tuple.Tuple2;
import org.apache.Flink.api.java.utils.ParameterTool;
import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.Flink.streaming.api.datastream.DataStream;
import org.apache.Flink.streaming.api.functions.ProcessFunction;
import org.apache.Flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.Flink.util.Collector;
import org.apache.Flink.streaming.connectors.Kafka.FlinkKafkaConsumer;
import org.apache.Flink.api.common.serialization.SimpleStringSchema;
import org.apache.Kafka.clients.consumer.ConsumerConfig;
import org.apache.Kafka.common.serialization.StringDeserializer;

import java.util.Properties;
import java.util.concurrent.CompletableFuture;

public class FlinkKafkaConsumerExample {
private static volatile int messageCount = 0;
private static volatile boolean shouldStop = false;
public static void main(String[] args) throws Exception {
// 设置执行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // Kafka 配置
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.146.132:9092"); // Kafka broker 地址
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); // 消费者组
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    // 创建 Kafka 消费者
    FlinkKafkaConsumer<String> KafkaConsumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
    KafkaConsumer.setStartFromEarliest(); // 从最早的消息开始消费
    DataStream<String> stream = env.addSource(KafkaConsumer);

    // 处理数据：分词和计数
    DataStream<Tuple2<String, Integer>> counts = stream
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);


    counts.addSink(new RichSinkFunction<Tuple2<String, Integer>>() {
        @Override
        public void invoke(Tuple2<String, Integer> value, Context context) {
            System.out.println(value);
            messageCount++;

            // 检查是否达到停止条件
            if (messageCount >= 2 && !shouldStop) {
                System.out.println("Processed 2 messages, stopping job.");
                shouldStop = true; // 设置标志位，表示应该停止
            }
        }
    });

    // 执行作业并获取 JobClient
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        try {
            // 启动作业并获取 JobClient
            org.apache.Flink.core.execution.JobClient jobClient = env.executeAsync("Flink Kafka WordCount");
            System.out.println("Job ID: " + jobClient.getJobID());

            // 监测条件并取消作业
            while (!shouldStop) {
                Thread.sleep(100); // 每100毫秒检查一次
            }

            // 达到停止条件时取消作业
            if (shouldStop) {
                System.out.println("Cancelling the job...");
                jobClient.cancel().get(); // 取消作业
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    });

    // 在主线程中等待作业结束
    future.join(); // 等待作业完成
}

// Tokenizer 类用于将输入字符串转化为单词
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
    @Override
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        String[] tokens = value.toLowerCase().split("\\W+");
        for (String token : tokens) {
            if (token.length() > 0) {
                out.collect(new Tuple2<>(token, 1));
            }
        }
    }
}

}
打包上传到Apache DolphinScheduler

新建Flink节点，并启动

在Apache DolphinScheduler的任务实例看启动日志：

在虚拟机启动生产者，输出字符串，然后可以在Flink查看输出Kafka生产的消息：

# linux # apache # flink
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

四天蒸发近2000亿，宇树科技股价近乎“腰斩”；苹果举办库克告别派对：9月1日正式卸任CEO；Anthropic两款新Claude模型曝光 | 极客头条

avatar AI编程社区
cover

Claude Code反超GitHub Copilot登顶第一、90%程序员已用上Agent，最新AI编码调查报告来了

avatar AI编程社区
cover

如何用好 Claude Code 写后端代码

avatar AI编程社区