配置好Flink的环境变量1、编辑环境变量sudo vim ~/.bashrc增加Flink的路径2、使环境变量生效#使环境变量生效 source ~/.bashrc #查看环境变量 echo $Flink_HOME修改Kafka、Flink以及DolphinScheduler的配置文件因为用的是虚拟机为了让外面的主机能够访问到虚拟机的网络需要修改下配置文件修改Kafka配置找到Kafka安装包下的config文件夹修改config下的server.properties文件修改listeners是为了外面的主机能够访问到虚拟机的Kafka还有把advertised.listeners改成虚拟机地址写样例的时候能连上虚拟机的Kafka地址不然默认连localhostbroker.id0 listenersPLAINTEXT://0.0.0.0:9092 #192.168.146.132修改成虚拟机ip advertised.listenersPLAINTEXT://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.xmlproject xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.example/groupId artifactIdFlink-Kafka-demo/artifactId version1.0-SNAPSHOT/version properties project.build.sourceEncodingUTF-8/project.build.sourceEncoding maven.compiler.source1.8/maven.compiler.source maven.compiler.target1.8/maven.compiler.target Flink.version1.18.1/Flink.version scala.binary.version2.12/scala.binary.version Kafka.version3.6.0/Kafka.version /properties dependencies !-- Flink核心依赖 -- dependency groupIdorg.apache.Flink/groupId artifactIdFlink-java/artifactId version${Flink.version}/version /dependency dependency groupIdorg.apache.Flink/groupId artifactIdFlink-streaming-java/artifactId version${Flink.version}/version /dependency dependency groupIdorg.apache.Flink/groupId artifactIdFlink-clients/artifactId version${Flink.version}/version /dependency !-- 连接器基础依赖 -- dependency groupIdorg.apache.Flink/groupId artifactIdFlink-connector-base/artifactId version${Flink.version}/version /dependency !-- Kafka连接器关键修改点 -- dependency groupIdorg.apache.Flink/groupId artifactIdFlink-connector-Kafka/artifactId version3.1.0-1.18/version /dependency dependency groupIdorg.apache.Kafka/groupId artifactIdKafka-clients/artifactId version${Kafka.version}/version /dependency !-- 日志依赖 -- dependency groupIdorg.slf4j/groupId artifactIdslf4j-simple/artifactId version1.7.36/version scoperuntime/scope /dependency /dependencies repositories repository idaliyun/id urlhttps://maven.aliyun.com/repository/public/url releases enabledtrue/enabled /releases snapshots enabledfalse/enabled /snapshots /repository repository idapache-releases/id urlhttps://repository.apache.org/content/repositories/releases//url /repository /repositories build plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-compiler-plugin/artifactId version3.8.1/version configuration source${maven.compiler.source}/source target${maven.compiler.target}/target /configuration /plugin plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goals goalshade/goal /goals configuration artifactSet excludes excludeorg.apache.Flink:force-shading/exclude excludecom.google.code.findbugs:jsr305/exclude excludeorg.slf4j:*/exclude /excludes /artifactSet filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters /configuration /execution /executions /plugin /plugins /build /projectFlinkKafkaConsumerExample.javaimport 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 消费者 FlinkKafkaConsumerString KafkaConsumer new FlinkKafkaConsumer(test, new SimpleStringSchema(), properties); KafkaConsumer.setStartFromEarliest(); // 从最早的消息开始消费 DataStreamString stream env.addSource(KafkaConsumer); // 处理数据分词和计数 DataStreamTuple2String, Integer counts stream .flatMap(new Tokenizer()) .keyBy(value - value.f0) .sum(1); counts.addSink(new RichSinkFunctionTuple2String, Integer() { Override public void invoke(Tuple2String, Integer value, Context context) { System.out.println(value); messageCount; // 检查是否达到停止条件 if (messageCount 2 !shouldStop) { System.out.println(Processed 2 messages, stopping job.); shouldStop true; // 设置标志位表示应该停止 } } }); // 执行作业并获取 JobClient CompletableFutureVoid 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 FlatMapFunctionString, Tuple2String, Integer { Override public void flatMap(String value, CollectorTuple2String, 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生产的消息