大数据之Flink（三）

9.3、转换算子
9.3.1、基本转换算子
9.3.1.1、映射map

一一映射

package transform;import bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @Title: MapDemo* @Author lizhe* @Package transform* @Date 2024/5/31 19:55* @description:*/
public class MapDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));SingleOutputStreamOperator<Object> map = sensorDataStreamSource.map((v) -> {return v.getId();});map.print();env.execute();}
}
9.3.1.2、过滤

转换操作，对数据流进行过滤，通过布尔条件表达式设置过滤条件，true正常输出，false被过滤掉

package transform;import bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @Title: MapDemo* @Author lizhe* @Package transform* @Date 2024/5/31 19:55* @description:* s1数据：一进一出* s2数据：一进二出* s3数据：一进零出*/
public class FilterDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s2", 2L, 2),new WaterSensor("s3", 3L, 3));SingleOutputStreamOperator<WaterSensor> filter = sensorDataStreamSource.filter((v) -> {return "s1".equals(v.getId());});filter.print();env.execute();}
}
9.3.1.3、扁平映射flatMap

将数据流中的整体拆分成个体使用。消费一个元素可产生多个元素。（一进多出）flatMap为flatten和map的结合，即按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。

package transform;import bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** @Title: MapDemo* @Author lizhe* @Package transform* @Date 2024/5/31 19:55* @description:*/
public class FlatMapDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 11),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));SingleOutputStreamOperator<String> flatmap = sensorDataStreamSource.flatMap(new FlatMapFunction<WaterSensor, String>() {@Overridepublic void flatMap(WaterSensor value, Collector<String> out) throws Exception {if ("s1".equals(value.getId())) {out.collect(String.valueOf(value.getVc()));} else if ("s2".equals(value.getId())) {out.collect(String.valueOf(value.getTs()));out.collect(String.valueOf(value.getVc()));}}});flatmap.print();env.execute();}
}

map使用的是return来控制一进一出，flatMap使用Collector，可调用多次采集器实现一进多出

9.3.1.4、聚合算子Aggregation

计算结果不仅依赖当前数据，还与之前的数据有关

  1. 按键分区keyby

    DataStream没有直接聚合的API。在flink中要聚合先进行可以不用keyby分区。keyby通过指定key将一条流划分成不同的分区，分区就是并行处理的子任务。

    package aggreagte;import bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** @Title: MapDemo* @Author lizhe* @Package transform* @Date 2024/5/31 19:55* @description:* s1数据：一进一出* s2数据：一进二出* s3数据：一进零出*/
public class KeybyDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 11),new WaterSensor("s1",11L,11),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));/** 按照id分组* 返回一个键控流KeyedStream，keyBy不是算子* keyby分组与分区的关系：* 1）keyby对数据进行分组，保证相同key的数据在同一个分区* 2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）* */KeyedStream<WaterSensor, String> keyBy = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor value) throws Exception {return value.getId();}});keyBy.print();env.execute();}
}

  2. 简单聚合

    按键分区后可以进行聚合操作，基本的API包括：sum、min、max、minBy、maxBy。

    sum

    package aggreagte;import bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SimpleAggDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 11),new WaterSensor("s1", 11L, 11),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));/** 按照id分组* 返回一个键控流KeyedStream，keyBy不是算子* keyby分组与分区的关系：* 1）keyby对数据进行分组，保证相同key的数据在同一个分区* 2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）* */KeyedStream<WaterSensor, String> keyBy = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor value) throws Exception {return value.getId();}});//传位置索引适用于tuple类型，不适合pojo类型
//        SingleOutputStreamOperator<WaterSensor> result = keyBy.sum(2);SingleOutputStreamOperator<WaterSensor> sum = keyBy.sum("vc");sum.print();//        SingleOutputStreamOperator<WaterSensor> max = keyBy.max("vc");
//        max.print();//        SingleOutputStreamOperator<WaterSensor> maxBy = keyBy.maxBy("vc");
//        maxBy.print();env.execute();}
}

    max

    package aggreagte;import bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SimpleAggDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s1", 11L, 11),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));/** 按照id分组* 返回一个键控流KeyedStream，keyBy不是算子* keyby分组与分区的关系：* 1）keyby对数据进行分组，保证相同key的数据在同一个分区* 2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）* */KeyedStream<WaterSensor, String> keyBy = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor value) throws Exception {return value.getId();}});//传位置索引适用于tuple类型，不适合pojo类型
//        SingleOutputStreamOperator<WaterSensor> result = keyBy.sum(2);
//        SingleOutputStreamOperator<WaterSensor> sum = keyBy.sum("vc");
//        sum.print();SingleOutputStreamOperator<WaterSensor> max = keyBy.max("vc");max.print();//        SingleOutputStreamOperator<WaterSensor> maxBy = keyBy.maxBy("vc");
//        maxBy.print();env.execute();}
}

    输出结果：

    WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=1, vc=11}
WaterSensor{id='s2', ts=2, vc=22}
WaterSensor{id='s3', ts=3, vc=3}

    maxby

    package aggreagte;import bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SimpleAggDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s1", 11L, 11),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));/** 按照id分组* 返回一个键控流KeyedStream，keyBy不是算子* keyby分组与分区的关系：* 1）keyby对数据进行分组，保证相同key的数据在同一个分区* 2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）* */KeyedStream<WaterSensor, String> keyBy = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor value) throws Exception {return value.getId();}});//传位置索引适用于tuple类型，不适合pojo类型
//        SingleOutputStreamOperator<WaterSensor> result = keyBy.sum(2);
//        SingleOutputStreamOperator<WaterSensor> sum = keyBy.sum("vc");
//        sum.print();//        SingleOutputStreamOperator<WaterSensor> max = keyBy.max("vc");
//        max.print();SingleOutputStreamOperator<WaterSensor> maxBy = keyBy.maxBy("vc");maxBy.print();env.execute();}
}

    输出结果

    WaterSensor{id='s1', ts=1, vc=1}
WaterSensor{id='s1', ts=11, vc=11}
WaterSensor{id='s2', ts=2, vc=22}
WaterSensor{id='s3', ts=3, vc=3}

    max与maxby对比（min与minby同理）：

    max只会取比较字段的最大值，非比较字段保留第一次的值

    maxby会取比较字段最大值这个对象

  3. 规约函数Reduce

    reduce：两两聚合，每个key第一条数据直接存起来并输出，聚合的结果作为下一次的第一条数据

    package aggreagte;import bean.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class ReduceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(new WaterSensor("s1", 1L, 1),new WaterSensor("s1", 11L, 11),new WaterSensor("s1", 22L, 22),new WaterSensor("s2", 2L, 22),new WaterSensor("s3", 3L, 3));/** 按照id分组* 返回一个键控流KeyedStream，keyBy不是算子* keyby分组与分区的关系：* 1）keyby对数据进行分组，保证相同key的数据在同一个分区* 2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）* */KeyedStream<WaterSensor, String> keyBy = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor value) throws Exception {return value.getId();}});SingleOutputStreamOperator<WaterSensor> reduce = keyBy.reduce(new ReduceFunction<WaterSensor>() {@Overridepublic WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {System.out.println("value1="+value1);System.out.println("value2="+value2);return new WaterSensor(value1.id, value2.ts, value1.vc + value2.vc);}});reduce.print();env.execute();}
}
9.3.1.5、自定义函数及分区

自定义函数分为：函数类、匿名函数、富函数类

物理分区即数据进入到多个线程中的哪个线程。常见分区策略：随机分配、轮询分配、重缩放、广播。

轮询（rebalance）：一般一个source对应一个kafka的partition，如果partition数据源不均匀（数据倾斜），可通过轮询分配进行负载均衡。

缩放（rescale）：实现轮询，局部组队，比rebalance高效。

广播（broadcast）：下发到下游所有子任务

9.3.1.6、分流

将一条数据拆分成完全独立的两条或多条流。基于一个DataStream通过筛选条件将符合条件的数据放到对应的流里。
在这里插入图片描述
只要针对同一条流进行多次独立调用filter()方法进行筛选就可以得到拆分之后的流，但是效率较低，所有数据都要过滤多次。

package split;import bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SplitByFilterDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStreamSource = env.socketTextStream("192.168.132.101", 7777);dataStreamSource.filter(value -> Integer.parseInt(value)%2==0).print("偶数流");dataStreamSource.filter(value -> Integer.parseInt(value) % 2 == 1).print("奇数流");env.execute();}
}

使用侧输出流实现分流，可实现数据筛选、告警等

  1. 使用process算子
  2. 定义OutputTag对象
  3. 调用ctx.output
  4. 通过主流获取侧输出流
package split;import bean.WaterSensor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;import java.lang.reflect.Type;public class SideOutputDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//如果是s1放到侧输出流s1中OutputTag<WaterSensor> s1 = new OutputTag<>("s1", Types.POJO(WaterSensor.class));//如果是s2放到侧输出流s2中OutputTag<WaterSensor> s2 = new OutputTag<>("s2", Types.POJO(WaterSensor.class));SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );SingleOutputStreamOperator<WaterSensor> process = dataStreamSource.process(new ProcessFunction<WaterSensor, WaterSensor>() {@Overridepublic void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {String id = value.getId();if (id.equals("s1")) {ctx.output(s1, value);} else if (id.equals("s2")) {ctx.output(s2, value);} else {//其他放到主流中out.collect(value);}}});//打印主输出流process.print("主输出流");//打印侧输出流process.getSideOutput(s1).print("s1侧输出流");process.getSideOutput(s2).print("s2侧输出流");env.execute();}
}
9.3.1.7、合流

1、联合union

最简单的合流操作就是将多条流合到一起，要求流中的数据类型必须相同，合并后新流包括所有流的元素，数据类型不变，一次可以合并多条流。
在这里插入图片描述

package combineDemo;import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class UnionDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3, 4, 5);DataStreamSource<Integer> source2 = env.fromElements(11, 22, 33, 44, 55);DataStreamSource<String> source3 = env.fromElements("111", "222","333","444","555");DataStream<Integer> union = source1.union(source2).union(source3.map(value -> Integer.parseInt(value)));union.print();env.execute();}
}

2、连接connect
为合并不同数据类型的数据flink提供connect合流操作。connect连接后得到的是ConnectedStream,形式上统一但内部内部各自数据形式不变，彼此之间相互独立。如要得到新的DataStream要使用“同处理”（co-process）,如map、flatmap等，各自处理各自的。

connect一次只能连接两条流。

package combineDemo;import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;public class ConnectDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3, 4, 5);DataStreamSource<String> source2 = env.fromElements("111", "222","333","444","555");ConnectedStreams<Integer, String> connect = source1.connect(source2);SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<Integer, String, String>() {@Overridepublic String map1(Integer value) throws Exception {return value.toString();}@Overridepublic String map2(String value) throws Exception {return value;}});map.print();env.execute();}
}

ConnectedStreams可以直接调用keyBy()进行按键分区得到的还是一个ConnectedStreams，通过keyBy()将两条流中key相同的数据放到了一起，然后针对来源的流再各自处理。（类似inner join）

package combineDemo;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;public class ConnectKeybyDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);DataStreamSource<Tuple2<Integer, String>> source1 = env.fromElements(Tuple2.of(1, "a1"),Tuple2.of(1, "a2"),Tuple2.of(2, "b"),Tuple2.of(3, "c"));DataStreamSource<Tuple3<Integer, String,Integer>> source2 = env.fromElements(Tuple3.of(1, "aa1",1),Tuple3.of(1, "aa2",2),Tuple3.of(2, "bb",1),Tuple3.of(3, "cc",1));ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> connect = source1.connect(source2);//多并行度下，要根据关联条件进行keyby，才能保证key相同的数据在一个子任务（线程）里，这样才能匹配上ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> keyBy = connect.keyBy(s1 -> s1.f0, s2 -> s2.f0);SingleOutputStreamOperator<String> process = keyBy.process(new CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>() {//每条流定一个hashmap用来存储数据Map<Integer, List<Tuple2<Integer, String>>> s1Cache = new HashMap<>();Map<Integer, List<Tuple3<Integer, String, Integer>>> s2Cache = new HashMap<>();@Overridepublic void processElement1(Tuple2<Integer, String> value, Context ctx, Collector<String> out) throws Exception {Integer id = value.f0;//先把s1数据存到map中if(!s1Cache.containsKey(id)){ArrayList<Tuple2<Integer, String>> s1Value = new ArrayList<>();s1Value.add(value);s1Cache.put(id, s1Value);}else {s1Cache.get(id).add(value);}if (s2Cache.containsKey(id)){for (Tuple3<Integer, String, Integer> s2Element : s2Cache.get(id)) {out.collect("s1"+value+"---------"+"s2"+s2Element);}}}@Overridepublic void processElement2(Tuple3<Integer, String, Integer> value, Context ctx, Collector<String> out) throws Exception {Integer id = value.f0;//先把s1数据存到map中if(!s2Cache.containsKey(id)){ArrayList<Tuple3<Integer, String, Integer>> s2Value = new ArrayList<>();s2Value.add(value);s2Cache.put(id, s2Value);}else {s2Cache.get(id).add(value);}if (s1Cache.containsKey(id)){for (Tuple2<Integer, String> s1Element : s1Cache.get(id)) {out.collect("s1"+s1Element+"---------"+"s2"+value);}}}});process.print();env.execute();}
}
9.4、输出算子sink

将计算结果写到外部存储
在这里插入图片描述
输出到外部系统参考官网。

9.4.1、输出到文件FileSink
package sink;import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.connector.sink.Sink;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.DefaultBucketFactoryImpl;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;import java.time.Duration;
import java.time.ZoneId;
import java.util.concurrent.TimeUnit;public class SinkFile {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//必须开启,否则文件一直是.inprogressenv.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);DataStreamSource<String> source = env.fromElements("111", "222","333","444","555");FileSink<String> sink = FileSink//官网示例.forRowFormat(new Path("f:/tmp"), new SimpleStringEncoder<String>("UTF-8")).withRollingPolicy(DefaultRollingPolicy.builder().withRolloverInterval(TimeUnit.MINUTES.toSeconds(5))
//                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)).withMaxPartSize(1024L).build()).build();//        FileSink<String> fileSink = FileSink
//                //输出行式存储文件
//                .<String>forRowFormat(new Path("f:/tmp"), new SimpleStringEncoder<>())
//                //输出文件配置
//                .withOutputFileConfig(
//                        OutputFileConfig.builder()
//                                .withPartPrefix("test")
//                                .withPartSuffix(".log")
//                                .build()
//                )
//                //文件分桶
//                .withBucketAssigner(new DateTimeBucketAssigner<>("yy-MM-dd", ZoneId.systemDefault()))
//                //文件滚动策略
//                .withRollingPolicy(DefaultRollingPolicy.builder()
//                        .withRolloverInterval(5L * 1000L)
//                        .withMaxPartSize(1L * 1024L)
//                        .build()
//                ).build();source.sinkTo(sink);env.execute();}
}
9.4.2、输出到kafka

参考官网：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/

package sink;import bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.table.KafkaSinkSemantic;
import org.apache.kafka.clients.producer.ProducerRecord;import java.nio.charset.StandardCharsets;
import java.util.Properties;public class SinkKafkaDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;Properties properties = new Properties();properties.setProperty("bootstrap.servers", "192.168.132.100:9092,192.168.132.101:9092,192.168.132.102:9092");SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );KafkaSerializationSchema<WaterSensor> serializationSchema = new KafkaSerializationSchema<WaterSensor>() {@Overridepublic ProducerRecord<byte[], byte[]> serialize(WaterSensor s,Long time  ) {return new ProducerRecord<>("test", // target topics.toString().getBytes(StandardCharsets.UTF_8)); // record contents}};dataStreamSource.addSink(new FlinkKafkaProducer<WaterSensor>("test",serializationSchema,properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE));env.execute();}
}![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/c12239189c82417f8d17f9f8312dcf97.png)##### 9.4.3、输出到MySQL参考官网：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/jdbc/```java
package sink;import bean.WaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;public class SinkMysqlDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );dataStreamSource.print();SinkFunction<WaterSensor> waterSensorSinkFunction = JdbcSink.sink("insert into ws (id,ts,vc)  values (?, ?, ?)",                       // mandatorynew JdbcStatementBuilder<WaterSensor>() {@Overridepublic void accept(PreparedStatement preparedStatement, WaterSensor waterSensor) throws SQLException {preparedStatement.setString(1, waterSensor.id);preparedStatement.setLong(2, waterSensor.ts);preparedStatement.setInt(3, waterSensor.vc);}},                  // mandatoryJdbcExecutionOptions.builder().withBatchSize(1000).withBatchIntervalMs(200).withMaxRetries(5).build(),                  // optionalnew JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://localhost:3306/testflink?" +"autoReconnect=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai")
//                        .withDriverName("org.Mysql.Driver").withUsername("root").withPassword("123").withConnectionCheckTimeoutSeconds(60).build()                  // mandatory);dataStreamSource.addSink(waterSensorSinkFunction);env.execute();}
}

10、时间和窗口

窗口一般是划定的一段时间范围，即时间窗。窗口本事是截取有界数据的一种方式，对这个范围内的数据进行处理。

10.1、窗口分类
  1. 按照驱动类型分：时间窗口（定点发车）、计数窗口（人齐发车）
  2. 按照窗口分配数据的规则分：滚动窗口、滑动窗口、会话窗口、全局窗口
10.2、窗口API概述

按键分区和非按键分区

1、按键分区

按键分区后数据流被key分成多条逻辑流KeyedStream，窗口计算会在多个并行子任务上同时执行。相同key的数据会在一个子任务中，相当于每个key都定义了一组窗口各自独立进行统计计算。

2、非按键分区

原始流dataStreamSource不会分成多条逻辑流，窗口逻辑只能在一个任务上执行，相当于并行度为1。

10.3、窗口分配器

Window Assigners 是构建窗口算子的第一步，用来定义数据被分配到哪个窗口，即指定窗口的类型。一般使用.window()方法，传入Window Assigners参数，返回WindowedStream。非按键分区使用.windowAll()，返回AllWindowedStream.

基于时间：

  • 按键分区滚动窗口，窗口长度2秒

    keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(2)));

  • 按键分区滑动时间窗口，窗口长度10s，滑动步长2s

    keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(2)));

  • 按键分区会话窗口，窗口长度2s

    keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(2)));

基于计数：

  • 按键分区滚动窗口,窗口长度为5个元素

    keyedStream.countWindow(5);

  • 按键分区滑动窗口，窗口长度5个元素，滑动步长2个元素

    keyedStream.countWindow(5,2);
10.4、窗口函数

窗口分配器只收集数据，窗口函数Window Function进行计算操作。

各种流的相互关系
在这里插入图片描述

  • 增量聚合：来一条算一条，窗口触发时输出计算结果
  • 全窗口函数：数据来了不计算先存上，等窗口触发时计算并输出结果
10.4.1、增量聚合函数

每来一个数据就聚合一次

1、归约函数ReduceFunction

相同key的第一条数据来的时候不会调用reduce方法，来一条数据就算一条，窗口触发输出计算结果

package window;import bean.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class WindowAPIDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );KeyedStream<WaterSensor, String> keyedStream = dataStreamSource.keyBy(value -> value.getId());WindowedStream<WaterSensor, String, TimeWindow> windowStream = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));SingleOutputStreamOperator<WaterSensor> reduce = windowStream.reduce(new ReduceFunction<WaterSensor>() {@Overridepublic WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {return new WaterSensor(value1.id, value2.ts, value1.vc + value2.vc);}});reduce.print();env.execute();}
}

2、聚合函数Aggregate Function

ReduceFunction能解决大多归约聚合问题，但聚合状态类型、输出结果类型和输入数据类型必须一样。Aggregate Function更加灵活，有三种类型：输入IN、累加器ACC、输出OUT。输入IN是输入流中元素的数据类型；累加器类型ACC是聚合中间状态类型；输出OUT是最终计算结果类型。

  • 第一条数据来创建窗口和累加器
  • 增量聚合：来一条算一条（调用一次add方法）
  • 窗口输出调用一次getresult方法
  • 输入、输出、中间累加器的类型可以不一样
package window;import bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class WindowAggregateDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );KeyedStream<WaterSensor, String> keyedStream = dataStreamSource.keyBy(value -> value.getId());WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));SingleOutputStreamOperator<String> aggregate = window.aggregate(new AggregateFunction<WaterSensor, Integer, String>() {@Overridepublic Integer createAccumulator() {System.out.println("初始化累加器");return 0;}@Overridepublic Integer add(WaterSensor value, Integer accumulator) {System.out.println("调用add");return value.vc + accumulator;}@Overridepublic String getResult(Integer accumulator) {System.out.println("输出结果");return accumulator.toString();}@Overridepublic Integer merge(Integer a, Integer b) {//只有会话窗口才用return null;}});aggregate.print();env.execute();}
}
10.4.2、全窗口函数

1、窗口函数

.apply()，但是该方法能提供的上下文信息比较少，已经被ProcessWindowFunction全覆盖

window.apply(new WindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void apply(String key, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {}});

2、处理窗口函数

ProcessWindowFunction除了能拿到窗口数据外还能获取上下文对象。上下文包括窗口信息、当前的时间和状态信息（处理时间、事件时间水位线）

窗口触发时才调用一次，统一计算窗口内的所有数据

package window;import bean.WaterSensor;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;public class WindowProcessDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );KeyedStream<WaterSensor, String> keyedStream = dataStreamSource.keyBy(value -> value.getId());
//        dataStreamSource.windowAll();WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));SingleOutputStreamOperator<String> process = window.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {long start = context.window().getStart();long end = context.window().getEnd();String winStart = DateFormatUtils.format(start, "yyyy-MM-dd HH:mm:ss.SSS");String winEnd = DateFormatUtils.format(end, "yyyy-MM-dd HH:mm:ss.SSS");long l = elements.spliterator().estimateSize();out.collect("key=" + key + "的窗口[" + winStart + "," + winEnd + "]包含" + l + "条数据" + elements.toString());}});process.print();env.execute();}
}
10.4.3、增量聚合与全窗口函数接合使用

增量聚合Aggregate+全窗口的ProcessWindow

  1. 增量聚合函数处理数据：来一条算一条
  2. 窗口触发时，增量聚合结果（只有一条数据）传给全窗口函数
  3. 经过全窗口函数的处理后输出

从而实现了两者的优点（reduce函数也能传全窗口函数）

  1. 增量聚合：来一条算一条只存储中间计算结果，占用空间少
  2. 全窗口函数：可以通过上下文实现灵活的功能
package window;import bean.WaterSensor;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;public class WindowAggregateAndProcessDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();;env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> dataStreamSource = env.socketTextStream("192.168.132.101", 7777).map(value ->{String[] datas = value.split(",");return new WaterSensor(datas[0],Long.parseLong(datas[1]),Integer.parseInt(datas[2]) );} );KeyedStream<WaterSensor, String> keyedStream = dataStreamSource.keyBy(value -> value.getId());WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));SingleOutputStreamOperator<String> outputStreamOperator = window.aggregate(new MyAgg(), new MyProcess());outputStreamOperator.print();env.execute();}public static class MyAgg implements  AggregateFunction<WaterSensor, Integer, String>{@Overridepublic Integer createAccumulator() {System.out.println("初始化累加器");return 0;}@Overridepublic Integer add(WaterSensor value, Integer accumulator) {System.out.println("调用add");return value.vc + accumulator;}@Overridepublic String getResult(Integer accumulator) {System.out.println("输出结果");return accumulator.toString();}@Overridepublic Integer merge(Integer a, Integer b) {//只有会话窗口才用return null;}}public static class MyProcess extends ProcessWindowFunction<String, String, String, TimeWindow> {@Overridepublic void process(String key, Context context, Iterable<String> elements, Collector<String> out) throws Exception {long start = context.window().getStart();long end = context.window().getEnd();String winStart = DateFormatUtils.format(start, "yyyy-MM-dd HH:mm:ss.SSS");String winEnd = DateFormatUtils.format(end, "yyyy-MM-dd HH:mm:ss.SSS");long l = elements.spliterator().estimateSize();out.collect("key=" + key + "的窗口[" + winStart + "," + winEnd + "]包含" + l + "条数据" + elements.toString());}}
}
10.5、小结

触发器、移除器：现成的几个窗口都有默认的实现，一般不需要定义

以时间滚动窗口为例：

  • 窗口什么时候触发输出：时间进展>=窗口的最大时间戳（end-1ms）

  • 窗口是怎么划分的：start=取窗口长度的整数倍，向下取整，end=start+窗口长度，窗口左闭右开[start,end)

  • 窗口生命周期：

    创建：属于本窗口的第一条数据来的时候现new的，放入一个singleton单例的集合中；

    销毁（关窗）：时间进展>=窗口的最大时间戳（end-1ms）+允许迟到时间（默认为0）

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/422341.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

创建Java项目，可实现main方法运行，实现对性能数据的处理

创建Java项目，可实现main方法运行，实现对性能数据的处理

1、Android Studio无法执行Java类的main方法问题及解决方法 Android Studio无法执行Java类的main方法问题及解决方法_delegatedbuild-CSDN博客 D:\workspaces\performanceTools\.idea 文件夹下&#xff0c;gardle.xml ,添加依赖 <option name"delegatedBuild"…
阅读更多...
WebGL系列教程二（环境搭建及着色器初始化）

WebGL系列教程二（环境搭建及着色器初始化）

目录 1 前言2 新建html页面3 着色器介绍3.1 顶点着色器、片元着色器与光栅化的概念3.2 声明顶点着色器3.3 声明片元着色器 4 坐标系(右手系)介绍5 着色器初始化5.1 给一个画布canvas5.2 获取WebGL对象5.3 创建着色器对象5.4 获取着色器对象的源5.5 绑定着色器的源5.6 编译着色器…
阅读更多...
对称矩阵的压缩存储

对称矩阵的压缩存储

1.给自己出题&#xff1a;自己动手创造&#xff0c;画一个5行5列的对称矩阵 2.画图&#xff1a;按“行优先”压缩存储上述矩阵&#xff0c;画出一维数组的样子 3.简答&#xff1a;写出元素 i,j 与 数组下标之间的对应关系 4.画图&#xff1a;按“列优先”压缩存储上述矩阵&a…
阅读更多...
接口测试用例的编写

接口测试用例的编写

&#x1f345; 点击文末小卡片&#xff0c;免费获取软件测试全套资料&#xff0c;资料在手&#xff0c;涨薪更快 1、接口测试发现的典型问题 接口测试经常遇到的bug和问题&#xff0c;如下&#xff1a; 传入参数处理不当&#xff0c;导致程序crash类型溢出&#xff0c;导…
阅读更多...
Docker部署MySQL8.0.39报错解决方案

Docker部署MySQL8.0.39报错解决方案

Docker部署MySQL8.0.39报错解决方案 2024-09-11T06:09:09.317582Z 0 [Warning] [MY-010139] [Server] Changed limits: max_open_files: 1024 (requested 8161) 2024-09-11T06:09:09.317586Z 0 [Warning] [MY-010142] [Server] Changed limits: table_open_cache: 431 (reques…
阅读更多...
GUI编程09：鼠标监听事件、模拟画图工具

GUI编程09：鼠标监听事件、模拟画图工具

视频链接&#xff1a;11、鼠标监听事件、模拟画图工具_哔哩哔哩_bilibilihttps://www.bilibili.com/video/BV1DJ411B75F?p11&vd_sourceb5775c3a4ea16a5306db9c7c1c1486b5 模拟画图工具的实现逻辑图&#xff1a; 实现代码&#xff1a; package com.yundait.lesson03;impo…
阅读更多...
mysql学习教程，从入门到精通，SQL AND OR 运算符（12）

mysql学习教程，从入门到精通，SQL AND OR 运算符（12）

1、SQL AND & OR 运算符 在本教程中&#xff0c;您将学习如何在子句中使用ASELECT column1_name, column2_name, columnN_nameFROM table_nameWHERE condition1 AND condition2;ND&#xff06;OR运算符&#xff0c;WHERE以根据多个条件过滤记录。 1.1、根据条件选择记录 …
阅读更多...
Linux: network: esp:收到了重复的包？

Linux: network: esp:收到了重复的包？

最近遇到一个问题，是说收到了dup的ESP包，这是表象上的两个相同的ESP，那是因为在wireshark的首选项里IPv4，没有选择重组分片包，导致wireshark先做了ESP的解析，如果选择IPv4协议里的重组分片包，会看到下面说有三个分片，而且其中一个有overlap。 所以，这个目前还是未解的…
阅读更多...
动手学深度学习（三）深度学习计算

动手学深度学习（三）深度学习计算

一、模型构造 1、继承Module类来构造模型来构造模型 class MLP(nn.Module):# 声明带有模型参数的层&#xff0c;这里声明了两个全连接层def __init__(self, **kwargs):# 调用MLP父类Block的构造函数来进行必要的初始化。这样在构造实例时还可以指定其他函数# 参数&#xff0c…
阅读更多...
[数据集][目标检测]汽车头部尾部检测数据集VOC+YOLO格式5319张3类别

[数据集][目标检测]汽车头部尾部检测数据集VOC+YOLO格式5319张3类别

数据集制作单位&#xff1a;未来自主研究中心(FIRC) 版权单位&#xff1a;未来自主研究中心(FIRC) 版权声明&#xff1a;数据集仅仅供个人使用&#xff0c;不得在未授权情况下挂淘宝、咸鱼等交易网站公开售卖,由此引发的法律责任需自行承担 数据集格式&#xff1a;Pascal VOC格…
阅读更多...
需求分析概述

需求分析概述

为什么要进行需求分析呢&#xff1f; 笑话&#xff1a;富翁娶妻 某富翁想要娶老婆&#xff0c;有三个人选&#xff0c;富翁给了三个女孩各一千元&#xff0c;请 她们把房间装满。第一个女孩买了很多棉花&#xff0c;装满房间的1/2。第 二个女孩买了很多气球&#xff0c;装满…
阅读更多...
Java多线程（一）

Java多线程（一）

目录 Java多线程&#xff08;一&#xff09; 线程与进程基本介绍 并发和并行基本介绍 CPU调度基本介绍 主线程基本介绍 创建线程对象与相关方法 继承Thread类创建线程对象 多线程在内存中运行的原理 Thread类中常用的方法 Thread类中关于线程优先级的方法 守护线程与Thread类中…
阅读更多...
Kafka【十三】消费者消费消息的偏移量

Kafka【十三】消费者消费消息的偏移量

偏移量offset是消费者消费数据的一个非常重要的属性。默认情况下&#xff0c;消费者如果不指定消费主题数据的偏移量&#xff0c;那么消费者启动消费时&#xff0c;无论当前主题之前存储了多少历史数据&#xff0c;消费者只能从连接成功后当前主题最新的数据偏移位置读取&#…
阅读更多...
信息安全数学基础（8）整数分解

信息安全数学基础（8）整数分解

前言 在信息安全数学基础中&#xff0c;整数分解是一个核心概念&#xff0c;它指的是将一个正整数表示为几个正整数的乘积的形式。虽然对于任何正整数&#xff0c;理论上都可以进行分解&#xff08;除了1只能分解为1本身&#xff09;&#xff0c;但整数分解在密码学和信息安全中…
阅读更多...
实战千问2大模型第三天——Qwen2-VL-7B（多模态）视频检测和批处理代码测试

实战千问2大模型第三天——Qwen2-VL-7B（多模态）视频检测和批处理代码测试

画面描述：这个视频中，一位穿着蓝色西装的女性站在室内，背景中可以看到一些装饰品和植物。她双手交叉放在身前，面带微笑，似乎在进行一场演讲或主持活动。她的服装整洁，显得非常专业和自信。 一、简介 阿里通义千问开源新一代视觉语言模型Qwen2-VL。其中，Qwen2-VL-72B在大…
阅读更多...
使用虚拟信用卡WildCard轻松订阅POE：全面解析平台功能与订阅方式

使用虚拟信用卡WildCard轻松订阅POE：全面解析平台功能与订阅方式

POE&#xff08;Platform of Engagement&#xff09;是一个由Quora推出的人工智能聊天平台&#xff0c;汇集了多个强大的AI聊天机器人&#xff0c;如GPT-4、Claude、Sage等。POE提供了一个简洁、统一的界面&#xff0c;让用户能够便捷地与不同的AI聊天模型进行互动。这种平台的…
阅读更多...
先攒一波硬件，过几年再给电脑升级，靠谱吗？想啥呢？

先攒一波硬件，过几年再给电脑升级，靠谱吗？想啥呢？

前言 最近有小伙伴发来消息&#xff1a;我可以今年先买电脑的部分硬件&#xff0c;明年再买电脑的另一部分硬件&#xff0c;再组装起来不就是一台电脑了吗&#xff1f; 这确实是一个很好的办法。 我还记得大学有个室友&#xff0c;从大一每个月省吃俭用&#xff0c;攒下的钱…
阅读更多...
Linux学习笔记(黑马程序员，前四章节)

Linux学习笔记(黑马程序员，前四章节)

第一章 快照 虚拟机快照&#xff1a; 通俗来说&#xff0c;在学习阶段我们无法避免的可能损坏Linux操作系统&#xff0c;如果损坏的话&#xff0c;重新安装一个Linux操作系统就会十分麻烦。VMware虚拟机支持为虚拟机制作快照。通过快照将当前虚拟机的状态保存下来&#xff0c;…
阅读更多...
力扣100题——贪心算法

力扣100题——贪心算法

概述 贪心算法&#xff08;Greedy Algorithm&#xff09;是一种在解决问题时&#xff0c;按照某种标准在每一步都选择当前最优解&#xff08;局部最优解&#xff09;的算法。它期望通过一系列局部最优解的选择&#xff0c;最终能够得到全局最优解。 贪心算法的核心思想 贪心算…
阅读更多...
Springboot中自定义监听器

Springboot中自定义监听器

一、监听器模式图 二、监听器三要素 广播器&#xff1a;用来发布事件 事件&#xff1a;需要被传播的消息 监听器&#xff1a;一个对象对一个事件的发生做出反应&#xff0c;这个对象就是事件监听器 三、监听器的实现方式 1、实现自定义事件 自定义事件需要继承ApplicationEv…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023