代码题会是什么东西好慌 先把章末习题总结一波看看实力
实验一 Flink集群安装和初步使用
是的期末了我连环境都还没配 懒得配Linux了 用Maven速成一下子 以下为模拟环境配置速成教程
安装JDK1.8 安装Jetbrain IDEA 新建项目这样点
然后里面这样填再然后点创建
再然后pom.xml文件改成下面这样 改完右上角有个小按钮点一下自动下载一下依赖就行了
<?xml version="1.0" encoding="UTF-8"?><projectxmlns="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>com.example</groupId><artifactId>flink-demo</artifactId><version>1.0-SNAPSHOT</version><!-- 核心配置：JDK版本、Flink版本、Scala版本（确保依赖匹配） --><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><flink.version>1.14.5</flink.version><!-- 已验证依赖存在的稳定版本 --><scala.binary.version>2.12</scala.binary.version><!-- 与Flink 1.14.5匹配的Scala版本 --><slf4j.version>1.7.36</slf4j.version><!-- 日志依赖版本（兼容Flink） --></properties><dependencies><!-- 1. Flink流处理核心依赖（必须） --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!-- 2. Flink WebUI依赖（免安装可视化核心，已验证存在） --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime-web_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!-- 3. 日志依赖（避免运行时报日志相关错误） --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>${slf4j.version}</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>${slf4j.version}</version></dependency></dependencies><!-- 打包插件（可选，后续需要打包运行时可用） --><build><plugins><!-- Maven编译插件（确保用JDK 1.8编译） --><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><encoding>UTF-8</encoding></configuration></plugin><!-- Shade插件（打包时包含所有依赖，避免运行时缺包） --><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:flink-shaded-force-shading</exclude></excludes></artifactSet><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>com.example.FlinkHelloWorld</mainClass><!-- 你的主类全路径 --></transformer></transformers></configuration></execution></executions></plugin></plugins></build></project>实验二 使用多种算子完成Flink文本处理实验
因为我们没虚拟机所以要先配nc模拟流数据输入 下载这个然后配环境变量 之后流输入的内容我们就开个cmd然后输入nc -lk 7777之后需要传入什么数据直接继续往cmd后面输入就行了
然后直接写 运行 胜利结算
packageorg.example;importorg.apache.flink.api.common.functions.FilterFunction;importorg.apache.flink.api.common.functions.ReduceFunction;importorg.apache.flink.api.java.functions.KeySelector;importorg.apache.flink.streaming.api.datastream.KeyedStream;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.common.functions.MapFunction;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.util.Collector;publicclassWordCountJob{publicstaticvoidmain(String[]args)throwsException{// 设置Flink执行环境finalStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();// 创建一个socket数据源DataStream<String>text=env.socketTextStream("localhost",7777);// 将文本转换为小写DataStream<String>words=text.map(newMapFunction<String,String>(){@OverridepublicStringmap(Stringvalue)throwsException{returnvalue.toLowerCase();}});// 分割单词并为每个单词附加一个计数器为1的TupleSingleOutputStreamOperator<Tuple2<String,Integer>>wordAndOne=words.flatMap(newFlatMapFunction<String,Tuple2<String,Integer>>(){@OverridepublicvoidflatMap(Stringvalue,Collector<Tuple2<String,Integer>>out)throwsException{String[]words=value.split(" ");for(Stringword:words){Tuple2<String,Integer>wordsAndOne=Tuple2.of(word,1);out.collect(wordsAndOne);}}});// 过滤包含字母"a"的单词wordAndOne=wordAndOne.filter(newFilterFunction<Tuple2<String,Integer>>(){@Overridepublicbooleanfilter(Tuple2<String,Integer>stringIntegerTuple2)throwsException{returnstringIntegerTuple2.f0.contains("a");}});// 按单词进行分组，创建一个KeyedStreamKeyedStream<Tuple2<String,Integer>,String>wordAndOneKS=wordAndOne.keyBy(newKeySelector<Tuple2<String,Integer>,String>(){@OverridepublicStringgetKey(Tuple2<String,Integer>value)throwsException{returnvalue.f0;}});// 对每个单词的计数进行累加SingleOutputStreamOperator<Tuple2<String,Integer>>sumDS=wordAndOneKS.reduce(newReduceFunction<Tuple2<String,Integer>>(){@OverridepublicTuple2<String,Integer>reduce(Tuple2<String,Integer>value1,Tuple2<String,Integer>value2)throwsException{Stringword=value1.f0;intcount=value1.f1+value2.f1;returnTuple2.of(word,count);}});// 打印结果sumDS.print();// 执行任务env.execute("WordCountFlinkJob");}}
)
)
)
)
