大数据之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.mzph.cn/news/879172.shtml

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

相关文章

鸿蒙交互事件开发04——手势事件

鸿蒙交互事件开发04——手势事件

1 概 述 手势事件是移动应用开发中最常见的事件之一&#xff0c;鸿蒙提供了一些方法来绑定手势事件。通过给各个组件绑定不同的手势事件&#xff0c;并设计事件的响应方式&#xff0c;当手势识别成功时&#xff0c;ArkUI框架将通过事件回调通知组件手势识别的结果。 …
阅读更多...
王道考研操作系统笔记(一)

王道考研操作系统笔记(一)

虚拟内存的定义和特征&#xff1a; 基于局部性的原理&#xff0c; 在程序装入时&#xff0c;可以将程序中很快用到的部分装入内存&#xff0c;暂时用不到的数据装入外存&#xff0c;就可以让程序开始执行&#xff0c;在程序执行过程中&#xff0c;当所访问的信息不在内存的时…
阅读更多...
frida主动调用init_array中的函数

frida主动调用init_array中的函数

ida打开目标so&#xff0c;查看要主动调用的函数 前提是先过掉检测frida等等&#xff0c;然后控制台启动 输出so地址 Process.findModuleByName("libmod.so") New函数 var aa new NativeFunction(ptr(0x785e002000).add(0x134EC0),"void",[]) 主动调用 a…
阅读更多...
如何让人工智能训练更快

如何让人工智能训练更快

影响人工智能训练时间的因素 在深度学习训练中&#xff0c;训练时间的计算涉及到多个因素&#xff0c;包括 epoch 数、全局 batch size、微 batch size、计算设备数量等。下面是一个基本的公式来说明这些参数之间的关系&#xff08;注意&#xff0c;这只是一个基本的说明公式&…
阅读更多...
Makefile文件理解

Makefile文件理解

https://zhuanlan.zhihu.com/p/629855009 参考链接 这个链接我没都看&#xff0c;等用的时候再看吧 我遇到的文件是下面这张图片&#xff0c;然后23行两条命令和在命令行中执行是一样的。
阅读更多...
E32.【C语言 】练习:蓝桥杯题 懒羊羊字符串

E32.【C语言 】练习:蓝桥杯题 懒羊羊字符串

1.题目 【问题描述】 “懒羊羊”字符串是一种特定类型的字符串&#xff0c;它由三个字符组成&#xff0c;具有以下特点: 1.字符串长度为 3. 2.包含两种不同的字母。 3.第二个字符和第三个字符相同 换句话说&#xff0c;“懒羊羊”字符串的形式应为 ABB&#xff0c;其中A和B是不…
阅读更多...
k8s 资源管理

k8s 资源管理

文章目录 ResourceQuota什么是资源配额定义一个ResourceQuotaResourceQuota的使用 LimitRangeLimitRange的用途示例1&#xff1a;配置默认的requests和limits 节点故障大部分都是由于资源分配不合理、超额分配引起的&#xff0c;因此需要用某个技术手段保证节点的资源不会过大地…
阅读更多...
【HCIA-Datacom】网络参考模型

【HCIA-Datacom】网络参考模型

网络参考模型 1. 应用和数据2. 网络参考模型与标准协议OSI参考模型TCP/IP参考模型与标准协议常见的几种协议应用层传输层网络层和数据链路层物理层 3. 数据通信过程思考题测一测 ⭐在上一章节中已经给大家介绍了我们网络与生活之间的一些联系 ⭐这一章节主要学习的内容叫做网络…
阅读更多...
GDB 查看汇编

GDB 查看汇编

查看汇编 x disassemble
阅读更多...
C++从入门到起飞之——string类用法 全方位剖析！

C++从入门到起飞之——string类用法 全方位剖析！

&#x1f308;个人主页&#xff1a;秋风起&#xff0c;再归来~&#x1f525;系列专栏&#xff1a;C从入门到起飞 &#x1f516;克心守己&#xff0c;律己则安 目录 1. 为什么学习string类&#xff1f; 1.1 C语言中的字符串 1.2 两个面试题(先不做讲解) 2. 标准库中…
阅读更多...
Vulnhub-RickdiculouslyEasy靶机攻略

Vulnhub-RickdiculouslyEasy靶机攻略

御剑扫描到ip 一.第一个flag 主机扫描 目录扫描 二.网页信息收集-第二个flag 9090也开放了web服务所以我们在IP地址后面加端口试试&#xff0c;如下图&#xff0c;加上了端口&#xff0c;并且发现了第二个flag&#xff0c;也对第二个flag进行了简单的探索也没有发现什么可以…
阅读更多...
内存卡提示格式化怎么办？轻松应对格式化

内存卡提示格式化怎么办？轻松应对格式化

在日常使用电子设备时&#xff0c;我们有时会遇到内存卡提示需要格式化的情况。这种情况往往让人感到焦虑和困惑&#xff0c;因为格式化意味着内存卡上的所有数据都将被清除。然而&#xff0c;在遇到这种情况时&#xff0c;我们不必过于惊慌。本文将介绍四种方法&#xff0c;帮…
阅读更多...
如何获取MySQL数据表的列信息

如何获取MySQL数据表的列信息

在数据库管理中&#xff0c;了解表的结构是至关重要的。在MySQL中&#xff0c;我们可以通过几种方式来获取数据表的列信息。这不仅可以帮助我们更好地理解表的结构&#xff0c;还可以在编写查询时提供便利。以下是三种常用的方法来获取MySQL数据表的列信息。 使用 SHOW COLUMN…
阅读更多...
UDP通信实现

UDP通信实现

目录 前言 一、基础知识 1、跨主机传输 1、字节序 2、主机字节序和网络字节序 3、IP转换 2、套接字 3、什么是UDP通信 二、如何实现UDP通信 1、socket():创建套接字 2、bind():绑定套接字 3、sendto():发送指定套接字文件数据 4、recvfrom():接收指定地址信息的数据 三…
阅读更多...
如何逆转Instagram账号流量减少？4个实用技巧分享

如何逆转Instagram账号流量减少？4个实用技巧分享

Instagram作为全球十大社媒之一&#xff0c;不仅是个人分享生活的平台&#xff0c;还是跨境卖家进行宣传推广和客户开发的关键工具。在运营Instagram的过程中&#xff0c;稍有不慎就容易出现账号被限流的情况&#xff0c;对于账号状态和运营工作的进行都十分不利。 一、如何判断…
阅读更多...
怎么调整图片亮度？关于调整图片亮度的几种方法推荐

怎么调整图片亮度？关于调整图片亮度的几种方法推荐

怎么调整图片亮度&#xff1f;调整图片亮度是图像编辑中的一项基本但至关重要的操作。亮度直接影响图像的视觉效果和清晰度&#xff0c;它能够改变照片的整体氛围和表现力。无论你是在处理个人拍摄的照片、编辑设计作品&#xff0c;还是进行专业的图像修复&#xff0c;适当的亮…
阅读更多...
API安全 | 发现API的5个小tips

API安全 | 发现API的5个小tips

在安全测试目标时&#xff0c;最有趣的测试部分是它的 API。API 是动态的&#xff0c;它们比应用程序的其他部分更新得更频繁&#xff0c;并且负责许多后端繁重的工作。在现代应用程序中&#xff0c;我们通常会看到 REST API&#xff0c;但也会看到其他形式&#xff0c;例如 Gr…
阅读更多...
adb有线连接正常，adb connect失败

adb有线连接正常，adb connect失败

adb connect失败 1. 确认两个设备在同一个局域网 2. 确认此网络是否有adb连接的权限(有的公司网络不允许adb) 3. 确认防火墙设置 如果前面3步都确认没问题&#xff0c;Ping ip也能成功&#xff0c;那么有可能就是端口的问题: step1&#xff1a; 先用有线连接设备&#xff0…
阅读更多...
macos系统内置php文件列表 系统自带php卸载方法

macos系统内置php文件列表 系统自带php卸载方法

在macos系统中, 自带已经安装了php, 根据不同的macos版本php的版本号可能不同, 我们可以通过 which php 命令来查看mac自带的默认php安装路径, 不过注意这个只是php的执行文件路径. 系统自带php文件列表 一下就是macos默认安装的php文件列表. macos 10.15内置PHP文件列表配置…
阅读更多...
java微信机器人制作教程

java微信机器人制作教程

Java实现微信小号做机器人 随着人工智能技术的发展&#xff0c;机器人在各行各业扮演着越来越重要的角色。在社交领域&#xff0c;微信机器人也逐渐受到人们的关注。本文将介绍如何使用Java实现一个简单的微信小号做机器人的功能。 常见开发功能&#xff1a; 好友管理&#…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023