Flink-DataStream API介绍(源算子、转换算子、输出算子)

news2024/11/11 5:19:30

文章目录

    • DataStream API(基础篇)
      • Flink 支持的数据类型
      • 执行环境(Execution Environment)
        • 创建执行环境
        • 执行模式(Execution Mode)
        • 触发程序执行
      • 源算子
        • 准备工作
        • 从集合中读取数据
        • 从文件读取数据
        • 从 Socket 读取数据
        • 从 Kafka 读取数据
        • 自定义 Source
      • 转换算子(Transformation)
        • 映射(map)
        • 过滤(filter)
        • 扁平映射(flatMap)
        • 聚合算子(Aggregation)
          • 按键分区(keyBy)
          • 简单聚合
          • 归约聚合(reduce)
        • 用户自定义函数(UDF)
        • 富函数类(Rich Function Classes)
        • 物理分区(Physical Partitioning)和分区策略
      • 输出算子(Sink)
        • 输出文件
        • 输出到 Kafka
        • 输出到 Redis
        • 输出到 Elasticsearch
        • 输出到 MySQL(JDBC)
        • 自定义 Sink 输出

DataStream API(基础篇)

Flink 有非常灵活的分层 API 设计,其中的核心层就是 DataStream/DataSet API。由于新版本已经实现了流批一体,DataSet API 将被弃用,官方推荐统一使用 DataStream API 处理流数据和批数据。

DataStream(数据流)本身是 Flink 中一个用来表示数据集合的类(Class),我们编写的Flink 代码其实就是基于这种数据类型的处理,所以这套核心 API 就以 DataStream 命名。

一个 Flink 程序,其实就是对 DataStream 的各种转换。具体来说,代码基本上都由以下几部分构成:

  • 获取执行环境(execution environment)
  • 读取数据源(source)
  • 定义基于数据的转换操作(transformations)
  • 定义计算结果的输出位置(sink)
  • 触发程序执行(execute)

Flink 支持的数据类型

(1)Flink 的类型系统

Flink 作为一个分布式处理框架,处理的是以数据对象作为元素的流。为了方便地处理数据,Flink 有自己一整套类型系统。Flink 使用“类型信息”(TypeInformation)来统一表示数据类型。TypeInformation 类是 Flink 中所有类型描述符的基类。它涵盖了类型的一些基本属性,并为每个数据类型生成特定的序列化器、反序列化器和比较器。

(2) Flink 支持的数据类型

简单来说,对于常见的 Java 和 Scala 数据类型,Flink 都是支持的。Flink 在内部,Flink对支持不同的类型进行了划分,这些类型可以在 Types 工具类中找到:

  • 基本类型

    所有 Java 基本类型及其包装类,再加上 Void、String、Date、BigDecimal 和 BigInteger。

  • 数组类型

    包括基本类型数组(PRIMITIVE_ARRAY)和对象数组(OBJECT_ARRAY)

  • 复合数据类型

    • Java 元组类型(TUPLE):这是 Flink 内置的元组类型,是 Java API 的一部分。最多25 个字段,也就是从 Tuple0~Tuple25,不支持空字段
    • Scala 样例类及 Scala 元组:不支持空字段
    • 行类型(ROW):可以认为是具有任意个字段的元组,并支持空字段
    • POJO:Flink 自定义的类似于 Java bean 模式的类
  • 辅助类型

    Option、Either、List、Map 等

  • 泛型类型(GENERIC)

    Flink 支持所有的 Java 类和 Scala 类。不过如果没有按照上面 POJO 类型的要求来定义,就会被 Flink 当作泛型类来处理Flink 会把泛型类型当作黑盒,无法获取它们内部的属性;它们也不是由 Flink 本身序列化的,而是由 Kryo 序列化的

在这些类型中,元组类型和 POJO 类型最为灵活,因为它们支持创建复杂类型。而相比之下,POJO 还支持在键(key)的定义中直接使用字段名,这会让我们的代码可读性大大增加。所以,在项目实践中,往往会将流处理程序中的元素类型定为 Flink 的 POJO 类型。

Flink 对 POJO 类型的要求如下:

  • 类是公共的(public)和独立的(standalone,也就是说没有非静态的内部类);
  • 类有一个公共的无参构造方法;
  • 类中的所有字段是 public 且非 final 的;或者有一个公共的 getter 和 setter 方法,这些方法需要符合 Java bean 的命名规范。

所以我们看到,之前的 UserBehavior,就是我们创建的符合 Flink POJO 定义的数据类型。

(3)类型提示(Type Hints)

Flink 还具有一个类型提取系统,可以分析函数的输入和返回类型,自动获取类型信息,从而获得对应的序列化器和反序列化器。但是,由于 Java 中泛型擦除的存在,在某些特殊情况下(比如 Lambda 表达式中),自动提取的信息是不够精细的——只告诉 Flink 当前的元素由“船头、船身、船尾”构成,根本无法重建出“大船”的模样;这时就需要显式地提供类型信息,才能使应用程序正常工作或提高其性能

为了解决这类问题,Java API 提供了专门的“类型提示”(type hints)。

回忆一下之前的 word count 流处理程序,我们在将 String 类型的每个词转换成(word,count)二元组后,就明确地用 returns 指定了返回的类型。因为对于 map 里传入的 Lambda 表达式,系统只能推断出返回的是 Tuple2 类型,而无法得到 Tuple2<String, Long>。只有显式地告诉系统当前的返回类型,才能正确地解析出完整数据。

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

这是一种比较简单的场景,二元组的两个元素都是基本数据类型。那如果元组中的一个元素又有泛型,该怎么处理呢?

Flink 专门提供了 TypeHint 类,它可以捕获泛型的类型信息,并且一直记录下来,为运行时提供足够的信息。我们同样可以通过.returns()方法,明确地指定转换之后的 DataStream 里元素的类型。

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})

执行环境(Execution Environment)

Flink 程序可以在各种上下文环境中运行:我们可以在本地 JVM 中执行程序,也可以提交到远程集群上运行。

不同的环境,代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时,首先必须获取当前 Flink 的运行环境,从而建立起与 Flink 框架之间的联系。只有获取了环境上下文信息,才能将具体的任务调度到不同的 TaskManager 执行。

创建执行环境

我 们 要 获 取 的 执 行 环 境 , 是StreamExecutionEnvironment 类的对象,这是所有 Flink 程序的基础。在代码中创建执行环境的方式,就是调用这个类的静态方法,具体有以下三种。

(1)getExecutionEnvironment

最简单的方式,就是直接调用 getExecutionEnvironment 方法。它会根据当前运行的上下文直接得到正确的结果:如果程序是独立运行的,就返回一个本地执行环境;如果是创建了 jar包,然后从命令行调用它并提交到集群执行,那么就返回集群的执行环境。也就是说,这个方法会根据当前运行的方式,自行决定该返回什么样的运行环境。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

(2)createLocalEnvironment

这个方法返回一个本地执行环境。可以在调用时传入一个参数,指定默认的并行度;如果不传入,则默认并行度就是本地的 CPU 核心数。

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();

(3)createRemoteEnvironment

这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号,并指定要在集群中运行的 Jar 包。

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
    .createRemoteEnvironment(
    "host", // JobManager 主机名
    1234, // JobManager 进程端口号
    "path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
);

在获取到程序执行环境后,我们还可以对执行环境进行灵活的设置。比如可以全局设置程序的并行度、禁用算子链,还可以定义程序的时间语义、配置容错机制。

执行模式(Execution Mode)

上面我们获取到的执行环境,是一个 StreamExecutionEnvironment,顾名思义它应该是做流处理的。那对于批处理,又应该怎么获取执行环境呢?

在之前的 Flink 版本中,批处理的执行环境与流处理类似,是调用类 ExecutionEnvironment的静态方法,返回它的对象:

// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

基于 ExecutionEnvironment 读入数据创建的数据集合,就是 DataSet;对应的调用的一整套转换方法,就是 DataSet API。

而从 1.12.0 版本起,Flink 实现了 API 上的流批统一。DataStream API 新增了一个重要特性:可以支持不同的“执行模式”(execution mode),通过简单的设置就可以让一段 Flink 程序在流处理和批处理之间切换。这样一来,DataSet API 也就没有存在的必要了。

  • 流执行模式(STREAMING)

    这是 DataStream API 最经典的模式,一般用于需要持续实时处理的无界数据流。默认情况下,程序使用的就是 STREAMING 执行模式。

  • 批执行模式(BATCH)

    专门用于批处理的执行模式, 这种模式下,Flink 处理作业的方式类似于 MapReduce 框架。对于不会持续计算的有界数据,我们用这种模式处理会更方便。

  • 自动模式(AUTOMATIC)

    在这种模式下,将由程序根据输入数据源是否有界,来自动选择执行模式。

BATCH 模式的配置方法

由于 Flink 程序默认是 STREAMING 模式,我们这里重点介绍一下 BATCH 模式的配置。主要有两种方式

  • 通过命令行配置,在提交作业时,增加 execution.runtime-mode 参数,指定值为 BATCH(推荐):

    bin/flink run -Dexecution.runtime-mode=BATCH ...
    
  • 通过代码配置,在代码中,直接基于执行环境调用 setRuntimeMode 方法,传入 BATCH 模式:

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setRuntimeMode(RuntimeExecutionMode.BATCH);
    

触发程序执行

有了执行环境,我们就可以构建程序的处理流程了:基于环境读取数据源,进而进行各种转换操作,最后输出结果到外部系统。

需要注意的是,写完输出(sink)操作并不代表程序已经结束。因为当 main()方法被调用时,其实只是定义了作业的每个执行操作,然后添加到数据流图中;这时并没有真正处理数据——因为数据可能还没来。Flink 是由事件驱动的,只有等到数据到来,才会触发真正的计算,这也被称为“延迟执行”或“懒执行”(lazy execution)。

所以我们需要显式地调用执行环境的 execute()方法,来触发程序执行。execute()方法将一直等待作业完成,然后返回一个执行结果(JobExecutionResult)。

源算子

创建环境之后,就可以构建数据处理的业务逻辑。想要处理数据,先得有数据,所以首要任务就是把数据读进来。

Flink 可以从各种来源获取数据,然后构建 DataStream 进行转换处理。Flink 代码中通用的添加 source 的方式,是调用执行环境的 addSource()方法:

DataStream<String> stream = env.addSource(...);

方法传入一个对象参数,需要实现 SourceFunction 接口;返回 DataStreamSource。这里的DataStreamSource 类继承自 SingleOutputStreamOperator 类,又进一步继承自 DataStream。所以很明显,读取数据的 source 操作是一个算子,得到的是一个数据流(DataStream)。

准备工作

为了更好地理解,我们先构建一个实际应用场景。比如网站的访问操作,可以抽象成一个三元组(用户名,用户访问的 urrl,用户访问 url 的时间戳),所以在这里,我们可以创建一个类 Event,将用户行为包装成它的一个对象。

import java.sql.Timestamp;
public class Event {
    public String user;
    public String url;
    public Long timestamp;
    public Event() {
    }
    public Event(String user, String url, Long timestamp) {
        this.user = user;
        this.url = url;
        this.timestamp = timestamp;
    }
    @Override
    public String toString() {
        return "Event{" +
            "user='" + user + '\'' +
            ", url='" + url + '\'' +
            ", timestamp=" + new Timestamp(timestamp) +
            '}';
    } }

这里需要注意,我们定义的 Event,有这样几个特点:

  • 类是公有(public)的
  • 有一个无参的构造方法
  • 所有属性都是公有(public)的
  • 所有属性的类型都是可以序列化的

Flink 会把这样的类作为一种特殊的 POJO 数据类型来对待,方便数据的解析和序列化。另外我们在类中还重写了 toString 方法,主要是为了测试输出显示更清晰。

从集合中读取数据

最简单的读取数据的方式,就是在代码中直接创建一个 Java 集合,然后调用执行环境的fromCollection 方法进行读取。这相当于将数据临时存储到内存中,形成特殊的数据结构后,作为数据源使用,一般用于测试。

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = 
        StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    ArrayList<Event> clicks = new ArrayList<>();
    clicks.add(new Event("Mary","./home",1000L));
    clicks.add(new Event("Bob","./cart",2000L));
    DataStream<Event> stream = env.fromCollection(clicks);
    stream.print();
    env.execute();
}

我们也可以不构建集合,直接将元素列举出来,调用 fromElements 方法进行读取数据:

DataStreamSource<Event> stream2 = env.fromElements(
    new Event("Mary", "./home", 1000L),
    new Event("Bob", "./cart", 2000L)
);

从文件读取数据

真正的实际应用中,自然不会直接将数据写在代码中。通常情况下,我们会从存储介质中获取数据,一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

DataStream<String> stream = env.readTextFile("clicks.csv");
  • 参数可以是目录,也可以是文件;

  • 路径可以是相对路径,也可以是绝对路径;

  • 相对路径是从系统属性 user.dir 获取路径: idea 下是 project 的根目录, standalone 模式下是集群节点根目录;

  • 也可以从 hdfs 目录下读取, 使用路径 hdfs://…, 由于 Flink 没有提供 hadoop 相关依赖, 需要 pom 中添加相关依赖:

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.7.5</version>
        <scope>provided</scope>
    </dependency>
    

从 Socket 读取数据

不论从集合还是文件,我们读取的其实都是有界数据。在流处理的场景中,数据往往是无界的。这时又从哪里读取呢?

一个简单的方式,就是我们之前用到的读取 socket 文本流。这种方式由于吞吐量小、稳定性较差,一般也是用于测试。

DataStream<String> stream = env.socketTextStream("localhost", 7777);

从 Kafka 读取数据

Kafka 作为分布式消息传输队列,是一个高吞吐、易于扩展的消息系统。而消息队列的传输方式,恰恰和流处理是完全一致的。所以可以说 Kafka 和 Flink 天生一对,是当前处理流式数据的双子星。在如今的实时流处理应用中,由 Kafka 进行数据的收集和传输,Flink 进行分析计算,这样的架构已经成为众多企业的首选,如图:

Flink官方提供了连接工具flink-connector-kafka,直接帮我们实现了一个消费者 FlinkKafkaConsumer,它就是用来读取 Kafka 数据的SourceFunction。

导入依赖:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

然后调用 env.addSource(),传入 FlinkKafkaConsumer 的对象实例就可以了。

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
public class SourceKafkaTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer", 
                               "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", 
                               "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");
        DataStreamSource<String> stream = env.addSource(new 
                                                        FlinkKafkaConsumer<String>(
                                                            "clicks",
                                                            new SimpleStringSchema(),
                                                            properties
                                                        ));
        stream.print("Kafka");
        env.execute();
    } 
}

创建 FlinkKafkaConsumer 时需要传入三个参数:

  • 第一个参数 topic,定义了从哪些主题中读取数据。可以是一个 topic,也可以是 topic列表,还可以是匹配所有想要读取的 topic 的正则表达式。当从多个 topic 中读取数据时,Kafka 连接器将会处理所有 topic 的分区,将这些分区的数据放到一条流中去。
  • 第二个参数是一个 DeserializationSchema 或者 KeyedDeserializationSchema。Kafka 消息被存储为原始的字节数据,所以需要反序列化成 Java 或者 Scala 对象。上面代码中使用的 SimpleStringSchema,是一个内置的 DeserializationSchema,它只是将字节数组简单地反序列化成字符串。DeserializationSchema 和 KeyedDeserializationSchema 是公共接口,所以我们也可以自定义反序列化逻辑。
  • 第三个参数是一个 Properties 对象,设置了 Kafka 客户端的一些属性。

自定义 Source

大多数情况下,前面的数据源已经能够满足需要。但是凡事总有例外,如果遇到特殊情况,我们想要读取的数据源来自某个外部系统,而 flink 既没有预实现的方法、也没有提供连接器,又该怎么办呢?

那就只好自定义实现 SourceFunction 了。

接下来我们创建一个自定义的数据源,实现 SourceFunction 接口。主要重写两个关键方法:run()和 cancel()。

  • run()方法:使用运行时上下文对象(SourceContext)向下游发送数据;
  • cancel()方法:通过标识位控制退出循环,来达到中断数据源的效果。

代码如下:

我们先来自定义一下数据源:

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Calendar;
import java.util.Random;
public class ClickSource implements SourceFunction<Event> {
    // 声明一个布尔变量,作为控制数据生成的标识位
    private Boolean running = true;
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        Random random = new Random(); // 在指定的数据集中随机选取数据
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", 
                         "./prod?id=2"};
        while (running) {
            ctx.collect(new Event(
                users[random.nextInt(users.length)],
                urls[random.nextInt(urls.length)],
                Calendar.getInstance().getTimeInMillis()
            ));
            // 隔 1 秒生成一个点击事件,方便观测
            Thread.sleep(1000);
        }
    }
    @Override
    public void cancel() {
        running = false;
    }
}

下面的代码我们来读取一下自定义的数据源。有了自定义的 source function,接下来只要调用 addSource()就可以了:

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SourceCustom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //有了自定义的 source function,调用 addSource 方法
        DataStreamSource<Event> stream = env.addSource(new ClickSource());
        stream.print("SourceCustom");
        env.execute();
    }
}

这里要注意的是 SourceFunction 接口定义的数据源,并行度只能设置为 1,如果数据源设置为大于 1 的并行度,则会抛出异常。如下程序所示:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;
public class SourceThrowException {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new ClickSource()).setParallelism(2).print();
        env.execute();
    } 
}

输出的异常如下:

Exception in thread "main" java.lang.IllegalArgumentException: The parallelism 
of non parallel operator must be 1.

所以如果我们想要自定义并行的数据源的话,需要使用 ParallelSourceFunction,示例程序如下:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import java.util.Random;
public class ParallelSourceExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new CustomSource()).setParallelism(2).print();
        env.execute();
    }
    public static class CustomSource implements ParallelSourceFunction<Integer> 
    {
        private boolean running = true;
        private Random random = new Random();
        @Override
        public void run(SourceContext<Integer> sourceContext) throws Exception {
            while (running) {
                sourceContext.collect(random.nextInt());
            }
        }
        @Override
        public void cancel() {
            running = false;
        }
    } 
}

输出结果如下:

2> -686169047
2> 429515397
2> -223516288
2> 1137907312
2> -380165730
2> 2082090389

转换算子(Transformation)

映射(map)

map 是大家非常熟悉的大数据操作算子,主要用于将数据流中的数据进行转换,形成新的数据流。简单来说,就是一个“一一映射”,消费一个元素就产出一个元素。

我们只需要基于 DataStrema 调用 map()方法就可以进行转换处理。方法需要传入的参数是接口 MapFunction 的实现;返回值类型还是 DataStream,不过泛型(流中的元素类型)可能改变。

public class TransMapTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L)
        );
        // 传入匿名类,实现 MapFunction
        stream.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event e) throws Exception {
                return e.user;
            }
        });
        // 传入 MapFunction 的实现类
        stream.map(new UserExtractor()).print();
        env.execute();
    }
    public static class UserExtractor implements MapFunction<Event, String> {
        @Override
        public String map(Event e) throws Exception {
            return e.user;
        }
    } }

在实现 MapFunction 接口的时候,需要指定两个泛型,分别是输入事件和输出事件的类型,还需要重写一个 map()方法,定义从一个输入事件转换为另一个输出事件的具体逻辑。

基于 DataStream 调用 map 方法,返回的其实是一个 SingleOutputStreamOperator。

public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper){}

这表示 map 是一个用户可以自定义的转换(transformation)算子,它作用于一条数据流上,转换处理的结果是一个确定的输出类型。当然,SingleOutputStreamOperator 类本身也继承自 DataStream 类,所以说 map 是将一个 DataStream 转换成另一个 DataStream 是完全正确的。

过滤(filter)

filter 转换操作,顾名思义是对数据流执行一个过滤,通过一个布尔条件表达式设置过滤条件,对于每一个流内元素进行判断,若为 true 则元素正常输出,若为 false 则元素被过滤掉。

进行 filter 转换之后的新数据流的数据类型与原数据流是相同的。filter 转换需要传入的参数需要实现 FilterFunction 接口,而 FilterFunction 内要实现 filter()方法,就相当于一个返回布尔类型的条件表达式。

public class TransFilterTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L)
        );
        // 传入匿名类实现 FilterFunction
        stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event e) throws Exception {
                return e.user.equals("Mary");
            }
        });
        // 传入 FilterFunction 实现类
        stream.filter(new UserFilter()).print();
        env.execute();
    }
    public static class UserFilter implements FilterFunction<Event> {
        @Override
        public boolean filter(Event e) throws Exception {
            return e.user.equals("Mary");
        }
    } 
}

扁平映射(flatMap)

flatMap 操作又称为扁平映射,主要是将数据流中的整体(一般是集合类型)拆分成一个一个的个体使用。消费一个元素,可以产生 0 到多个元素。flatMap 可以认为是“扁平化”(flatten)和“映射”(map)两步操作的结合,也就是先按照某种规则对数据进行打散拆分,再对拆分后的元素做转换处理。我们此前 WordCount 程序的第一步分词操作,就用到了flatMap。

public class TransFlatmapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L)
        );
        stream.flatMap(new MyFlatMap()).print();
        env.execute();
    }
    public static class MyFlatMap implements FlatMapFunction<Event, String> {
        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception 
        {
            if (value.user.equals("Mary")) {
                out.collect(value.user);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.url);
            }
        }
    }
}

聚合算子(Aggregation)

按键分区(keyBy)

对于 Flink 而言,DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合肯定要进行分区并行处理,这样才能提高效率。所以在 Flink 中,要做聚合,需要先进行分区;这个操作就是通过 keyBy 来完成的。

keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键(key),可以将一条流从逻辑上划分成不同的分区(partitions)。这里所说的分区,其实就是并行处理的子任务,也就对应着任务槽(task slot)。

基于不同的 key,流中的数据将被分配到不同的分区中去,这样一来,所有具有相同的 key 的数据,都将被发往同一个分区,那么下一步算子操作就将会在同一个 slot中进行处理了。

在内部,是通过计算 key 的哈希值(hash code),对分区数进行取模运算来实现的。所以这里 key 如果是 POJO 的话,必须要重写 hashCode()方法。

keyBy()方法需要传入一个参数,这个参数指定了一个或一组 key。有很多不同的方法来指定 key:

  • 比如对于 Tuple 数据类型,可以指定字段的位置或者多个位置的组合;
  • 对于 POJO 类型,可以指定字段的名称(String);
  • 另外,还可以传入 Lambda 表达式或者实现一个键选择器(KeySelector),用于说明从数据中提取 key 的逻辑。
public class TransKeyByTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L)
        );
        // 使用 Lambda 表达式
        KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user);
        // 使用匿名类实现 KeySelector
        KeyedStream<Event, String> keyedStream1 = stream.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event e) throws Exception {
                return e.user;
            }
        });
        env.execute();
    } 
}

需要注意的是,keyBy 得到的结果将不再是 DataStream,而是会将 DataStream 转换为KeyedStream。KeyedStream 可以认为是“分区流”或者“键控流”,它是对 DataStream 按照key 的一个逻辑分区,所以泛型有两个类型:除去当前流中的元素类型外,还需要指定 key 的类型。

KeyedStream 也继承自 DataStream,所以基于它的操作也都归属于 DataStream API。但它跟之前的转换操作得到的 SingleOutputStreamOperator 不同,只是一个流的分区操作,并不是一个转换算子。KeyedStream 是一个非常重要的数据结构,只有基于它才可以做后续的聚合操作(比如 sum,reduce);而且它可以将当前算子任务的状态(state)也按照 key 进行划分、限定为仅对当前 key 有效。

简单聚合

有了按键分区的数据流 KeyedStream,我们就可以基于它进行聚合操作了。Flink 为我们内置实现了一些最基本、最简单的聚合 API,主要有以下几种:

  • sum():在输入流上,对指定的字段做叠加求和的操作。
  • min():在输入流上,对指定的字段求最小值。
  • max():在输入流上,对指定的字段求最大值。
  • minBy():与 min()类似,在输入流上针对指定字段求最小值。不同的是,min()只计算指定字段的最小值,其他字段会保留最初第一个数据的值;而 minBy()则会返回包含字段最小值的整条数据。
  • maxBy():与 max()类似,在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。

简单聚合算子使用非常方便,语义也非常明确。这些聚合方法调用时,也需要传入参数;但并不像基本转换算子那样需要实现自定义函数,只要说明聚合指定的字段就可以了。指定字段的方式有两种:指定位置,和指定名称

对于元组类型的数据,同样也可以使用这两种方式来指定字段。需要注意的是,元组中字段的名称,是以 f0、f1、f2、…来命名的。

例如,下面就是对元组数据流进行聚合的测试:

public class TransTupleAggreationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple2<String, Integer>> stream = env.fromElements(
            Tuple2.of("a", 1),
            Tuple2.of("a", 3),
            Tuple2.of("b", 3),
            Tuple2.of("b", 4)
        );
        stream.keyBy(r -> r.f0).sum(1).print();
        stream.keyBy(r -> r.f0).sum("f1").print();
        stream.keyBy(r -> r.f0).max(1).print();
        stream.keyBy(r -> r.f0).max("f1").print();
        stream.keyBy(r -> r.f0).min(1).print();
        stream.keyBy(r -> r.f0).min("f1").print();
        stream.keyBy(r -> r.f0).maxBy(1).print();
        stream.keyBy(r -> r.f0).maxBy("f1").print();
        stream.keyBy(r -> r.f0).minBy(1).print();
        stream.keyBy(r -> r.f0).minBy("f1").print();
        env.execute();
    } 
}

而如果数据流的类型是 POJO 类,那么就只能通过字段名称来指定,不能通过位置来指定了。

public class TransPojoAggregationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L)
        );
        stream.keyBy(e -> e.user).max("timestamp").print(); // 指定字段名称
        env.execute();
    }
}

简单聚合算子返回的,同样是一个 SingleOutputStreamOperator,也就是从 KeyedStream 又转换成了常规的 DataStream。所以可以这样理解:keyBy 和聚合是成对出现的,先分区、后聚合,得到的依然是一个 DataStream。而且经过简单聚合之后的数据流,元素的数据类型保持不变。

归约聚合(reduce)

如果说简单聚合是对一些特定统计需求的实现,那么 reduce 算子就是一个一般化的聚合统计操作了。从大名鼎鼎的 MapReduce 开始,我们对 reduce 操作就不陌生:它可以对已有的数据进行归约处理,把每一个新输入的数据和当前已经归约出来的值,再做一个聚合计算。

与简单聚合类似,reduce 操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元素数据类型,所以输出类型和输入类型是一样的。

调用 KeyedStream 的 reduce 方法时,需要传入一个参数,实现 ReduceFunction 接口。接口在源码中的定义如下:

public interface ReduceFunction<T> extends Function, Serializable {
    T reduce(T value1, T value2) throws Exception;
}

ReduceFunction 接口里需要实现 reduce()方法,这个方法接收两个输入事件,经过转换处理之后输出一个相同类型的事件;所以,对于一组数据,我们可以先取两个进行合并,然后再将合并的结果看作一个数据、再跟后面的数据合并,最终会将它“简化”成唯一的一个数据,这也就是 reduce“归约”的含义。在流处理的底层实现过程中,实际上是将中间“合并的结果”作为任务的一个状态保存起来的;之后每来一个新的数据,就和之前的聚合状态进一步做归约。

我们可以单独定义一个函数类实现 ReduceFunction 接口,也可以直接传入一个匿名类。当然,同样也可以通过传入 Lambda 表达式实现类似的功能。

例子:我们将数据流按照用户 id 进行分区,然后用一个 reduce 算子实现 sum 的功能,统计每个用户访问的频次;进而将所有统计结果分到一组,用另一个 reduce 算子实现 maxBy 的功能,记录所有用户中访问频次最高的那个,也就是当前访问量最大的用户是谁。

public class TransReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 这里的 ClickSource()使用了之前自定义数据源小节中的 ClickSource()
        env.addSource(new ClickSource())
            // 将 Event 数据类型转换成元组类型
            .map(new MapFunction<Event, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(Event e) throws Exception {
                    return Tuple2.of(e.user, 1L);
                }
            })
            .keyBy(r -> r.f0) // 使用用户名来进行分流
            .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, 
                                                   Tuple2<String, Long> value2) throws Exception {
                    // 每到一条数据,用户 pv 的统计值加 1
                    return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                }
            })
            .keyBy(r -> true) // 为每一条数据分配同一个 key,将聚合结果发送到一条流中.reduce(new ReduceFunction<Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, 
                                                   Tuple2<String, Long> value2) throws Exception {
                    // 将累加器更新为当前最大的 pv 统计值,然后向下游发送累加器的值
                    return value1.f1 > value2.f1 ? value1 : value2;
                }
            })
            .print();
        env.execute();
    }
}

用户自定义函数(UDF)

这些接口有一个共同特点:全部都以算子操作名称 + Function 命名,例如:源算子需要实现 SourceFunction 接口,map 算子需要实现 MapFunction 接口,reduce 算子需要实现 ReduceFunction 接口。

而且查看源码会发现,它们都继承自 Function 接口;这个接口是空的,主要就是为了方便扩展为单一抽象方法(Single Abstract Method,SAM)接口,这就是我们所说的“函数接口”——比如 MapFunction 中需要实现一个 map()方法,ReductionFunction中需要实现一个 reduce()方法,它们都是 SAM 接口。我们知道,Java 8 新增的 Lambda 表达式就可以实现 SAM 接口;所以这样的好处就是,我们不仅可以通过自定义函数类或者匿名类来实现接口,也可以直接传入 Lambda 表达式。这就是所谓的用户自定义函数(user-defined function,UDF)。

//使用 map 函数也会出现类似问题,以下代码会报错
DataStream<Tuple2<String, Long>> stream3 = clicks.map( event -> Tuple2.of(event.user, 1L) );
stream3.print();

富函数类(Rich Function Classes)

“富函数类”也是 DataStream API 提供的一个函数类的接口,所有的 Flink 函数类都有其Rich 版本。富函数类一般是以抽象类的形式出现的。例如:RichMapFunction、RichFilterFunction、RichReduceFunction 等。

既然“富”,那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不同主要在于,富函数类可以获取运行环境的上下文,并拥有一些生命周期方法,所以可以实现更复杂的功能。

Rich Function 有生命周期的概念。典型的生命周期方法有:

  • open()方法,是 Rich Function 的初始化方法,也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map()或者 filter()方法被调用之前,open()会首先被调用。所以像文件 IO 的创建,数据库连接的创建,配置文件的读取等等这样一次性的工作,都适合在 open()方法中完成。
  • close()方法,是生命周期中的最后一个调用的方法,类似于解构方法。一般用来做一些清理工作。

需要注意的是,这里的生命周期方法,对于一个并行子任务来说只会调用一次;而对应的,·实际工作方法,例如 RichMapFunction 中的 map(),在每条数据到来后都会触发一次调用。

public class RichFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<Event> clicks = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L),
            new Event("Alice", "./prod?id=1", 5 * 1000L),
            new Event("Cary", "./home", 60 * 1000L)
        );
        // 将点击事件转换成长整型的时间戳输出
        clicks.map(new RichMapFunction<Event, Long>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println(" 索 引 为 " + 
                                   getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始");
            }
            @Override
            public Long map(Event value) throws Exception {
                return value.timestamp;
            }
            @Override
            public void close() throws Exception {
                super.close();
                System.out.println(" 索 引 为 " + 
                                   getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束");
            }
        })
            .print();
        env.execute();
    } 
}

一个常见的应用场景就是,如果我们希望连接到一个外部数据库进行读写操作,那么将连接操作放在 map()中显然不是个好选择——因为每来一条数据就会重新连接一次数据库;所以我们可以在 open()中建立连接,在 map()中读写数据,而在 close()中关闭连接。所以我们推荐的最佳实践如下:

public class MyFlatMap extends RichFlatMapFunction<IN, OUT>> {
    @Override
    public void open(Configuration configuration) {
        // 做一些初始化工作
        // 例如建立一个和 MySQL 的连接
    }
    @Override
    public void flatMap(IN in, Collector<OUT out) {
        // 对数据库进行读写
    }
    @Override
    public void close() {
        // 清理工作,关闭和 MySQL 数据库的连接。
    }
}

另外,富函数类提供了 getRuntimeContext()方法,可以获取到运行时上下文的一些信息,例如程序执行的并行度,任务名称,以及状态(state)。这使得我们可以大大扩展程序的功能,特别是对于状态的操作,使得 Flink 中的算子具备了处理复杂业务的能力。

物理分区(Physical Partitioning)和分区策略

前面介绍聚合算子时,已经提到了 keyBy,它就是一种按照键的哈希值来进行重新分区的操作。只不过这种分区操作只能保证把数据按key“分开”,至于分得均不均匀、每个 key 的数据具体会分到哪一区去,这些是完全无从控制的——所以我们有时也说,keyBy 是一种逻辑分区(logical partitioning)操作

如果说 keyBy 这种逻辑分区是一种“软分区”,那真正硬核的分区就应该是所谓的“物理分区”(physical partitioning)。也就是我们要真正控制分区策略,精准地调配数据,告诉每个数据到底去哪里。其实这种分区方式在一些情况下已经在发生了:例如我们编写的程序可能对多个处理任务设置了不同的并行度,那么当数据执行的上下游任务并行度变化时,数据就不应该还在当前分区以直通(forward)方式传输了:

  • 如果并行度变小,当前分区可能没有下游任务了;
  • 而如果并行度变大,所有数据还在原先的分区处理就会导致资源的浪费。所以这种情况下,系统会自动地将数据均匀地发往下游所有的并行任务,保证各个分区的负载均衡。

有些时候,我们还需要手动控制数据分区分配策略。比如当发生数据倾斜的时候,系统无法自动调整,这时就需要我们重新进行负载均衡,将数据流较为平均地发送到下游任务操作分区中去。Flink 对于经过转换操作之后的 DataStream,提供了一系列的底层操作接口,能够帮我们实现数据流的手动重分区。为了同 keyBy 相区别,我们把这些操作统称为“物理分区”操作。

物理分区与 keyBy 另一大区别在于,keyBy 之后得到的是一个 KeyedStream,而物理分区之后结果仍是 DataStream,且流中元素数据类型保持不变。从这一点也可以看出,分区算子并不对数据进行转换处理,只是定义了数据的传输方式。

常见的物理分区策略有随机分配(Random)、轮询分配(Round-Robin)、重缩放(Rescale)和广播(Broadcast),下边我们分别来做了解。

(1)随机分区(shuffle)

最简单的重分区方式就是直接“洗牌”。通过调用 DataStream 的.shuffle()方法,将数据随机地分配到下游算子的并行任务中。

我们可以做个简单测试:将数据读入之后直接打印到控制台,将输出的并行度设置为 4,中间经历一次 shuffle。执行多次,观察结果是否相同。

public class ShuffleTest {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 读取数据源,并行度为 1
        DataStreamSource<Event> stream = env.addSource(new ClickSource());
        // 经洗牌后打印输出,并行度为 4
        stream.shuffle().print("shuffle").setParallelism(4);
        env.execute();
    } 
}

输出:

shuffle:1> Event{user='Bob', url='./cart', timestamp=...}
shuffle:4> Event{user='Cary', url='./home', timestamp=...}
shuffle:3> Event{user='Alice', url='./fav', timestamp=...}
shuffle:4> Event{user='Cary', url='./cart', timestamp=...}
shuffle:3> Event{user='Cary', url='./fav', timestamp=...}
shuffle:1> Event{user='Cary', url='./home', timestamp=...}
shuffle:2> Event{user='Mary', url='./home', timestamp=...} shuffle:1> Event{user='Bob', url='./fav', timestamp=...}
shuffle:2> Event{user='Mary', url='./home', timestamp=...}
...

(2)轮询分区(Round-Robin)

轮询也是一种常见的重分区方式。简单来说就是“发牌”,按照先后顺序将数据做依次分发,如图所示。通过调用 DataStream 的.rebalance()方法,就可以实现轮询重分区。rebalance使用的是 Round-Robin 负载均衡算法,可以将输入流数据平均分配到下游的并行任务中去。

public class RebalanceTest {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 读取数据源,并行度为 1
        DataStreamSource<Event> stream = env.addSource(new ClickSource());
        // 经轮询重分区后打印输出,并行度为 4
        stream.rebalance().print("rebalance").setParallelism(4);
        env.execute();
    } 
}

(3)重缩放分区(rescale)

重缩放分区和轮询分区非常相似。当调用 rescale()方法时,其实底层也是使用 Round-Robin 算法进行轮询,但是只会将数据轮询发送到下游并行任务的一部分中,如图所示。也就是说,“发牌人”如果有多个,那么 rebalance 的方式是每个发牌人都面向所有人发牌;而 rescale的做法是分成小团体,发牌人只给自己团体内的所有人轮流发牌。

当下游任务(数据接收方)的数量是上游任务(数据发送方)数量的整数倍时,rescale的效率明显会更高。比如当上游任务数量是 2,下游任务数量是 6 时,上游任务其中一个分区的数据就将会平均分配到下游任务的 3 个分区中。

由于 rebalance 是所有分区数据的“重新平衡”,当 TaskManager 数据量较多时,这种跨节点的网络传输必然影响效率;而如果我们配置的 task slot 数量合适,用 rescale 的方式进行“局部重缩放”,就可以让数据只在当前 TaskManager 的多个 slot 之间重新分配,从而避免了网络传输带来的损耗。

public class RescaleTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 这里使用了并行数据源的富函数版本
        // 这样可以调用 getRuntimeContext 方法来获取运行时上下文的一些信息
        env.addSource(new RichParallelSourceFunction<Integer>() {
                @Override
                public void run(SourceContext<Integer> sourceContext) throws 
                    Exception {
                    for (int i = 0; i < 8; i++) {
                        // 将奇数发送到索引为 1 的并行子任务
                        // 将偶数发送到索引为 0 的并行子任务
                        if ((i + 1) % 2 == 
                            getRuntimeContext().getIndexOfThisSubtask()) {
                            sourceContext.collect(i + 1);
                        }
                    }
                }
                @Override
                public void cancel() {
                }
            })
            .setParallelism(2)
            .rescale()
            .print().setParallelism(4);
        env.execute();
    } 
}

这里使用 rescale 方法,来做数据的分区,输出结果是:

4> 3
3> 1
1> 2
1> 6
3> 5
4> 7
2> 4
2> 8

(4)广播(broadcast)

这种方式其实不应该叫做“重分区”,因为经过广播之后,数据会在不同的分区都保留一份,可能进行重复处理。可以通过调用 DataStream 的 broadcast()方法,将输入数据复制并发送到下游算子的所有并行任务中去。

public class BroadcastTest {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 读取数据源,并行度为 1
        DataStreamSource<Event> stream = env.addSource(new ClickSource());
        // 经广播后打印输出,并行度为 4
        stream. broadcast().print("broadcast").setParallelism(4);
        env.execute();
    } 
}

输出:

broadcast:3> Event{user='Mary', url='./cart', timestamp=...}
broadcast:1> Event{user='Mary', url='./cart', timestamp=...}
broadcast:4> Event{user='Mary', url='./cart', timestamp=...}
broadcast:2> Event{user='Mary', url='./cart', timestamp=...}
broadcast:2> Event{user='Alice', url='./fav', timestamp=...}
broadcast:1> Event{user='Alice', url='./fav', timestamp=...}
broadcast:3> Event{user='Alice', url='./fav', timestamp=...}
broadcast:4> Event{user='Alice', url='./fav', timestamp=...}

(5)全局分区(global)

全局分区也是一种特殊的分区方式。这种做法非常极端,通过调用.global()方法,会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了 1,所以使用这个操作需要非常谨慎,可能对程序造成很大的压力。

(6)自定义分区(Custom)

当 Flink 提 供 的 所 有 分 区 策 略 都 不 能 满 足 用 户 的 需 求 时 , 我 们 可 以 通 过 使 用partitionCustom()方法来自定义分区策略。在调用时,方法需要传入两个参数,第一个是自定义分区器(Partitioner)对象,第二个是应用分区器的字段,它的指定方式与 keyBy 指定 key 基本一样:可以通过字段名称指定,也可以通过字段位置索引来指定,还可以实现一个 KeySelector。

例如,我们可以对一组自然数按照奇偶性进行重分区。代码如下:

public class CustomPartitionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 将自然数按照奇偶分区
        env.fromElements(1, 2, 3, 4, 5, 6, 7, 8)
            .partitionCustom(new Partitioner<Integer>() {
                @Override
                public int partition(Integer key, int numPartitions) {
                    return key % 2;
                }
            }, new KeySelector<Integer, Integer>() {
                @Override
                public Integer getKey(Integer value) throws Exception {
                    return value;
                }
            })
            .print().setParallelism(2);
        env.execute();
    } 
}

输出算子(Sink)

Flink 的 DataStream API 专门提供了向外部写入数据的方法:addSink。与 addSource 类似,addSink 方法对应着一个“Sink”算子,主要就是用来实现与外部系统连接、并将数据提交写入的;Flink 程序中所有对外的输出操作,一般都是利用 Sink 算 子完成的。

之前我们一直在使用的 print 方法其实就是一种 Sink,它表示将数据流写入标准控制台打印输出。查看源码可以发现,print 方法返回的就是一个 DataStreamSink。

public DataStreamSink<T> print(String sinkIdentifier) {
    PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier, false);
    return addSink(printFunction).name("Print to Std. Out");
}

与 Source 算子非常类似,除去一些 Flink 预实现的 Sink,一般情况下 Sink 算子的创建是通过调用 DataStream 的.addSink()方法实现的:

stream.addSink(new SinkFunction());

addSource 的参数需要实现一个 SourceFunction 接口;类似地,addSink 方法同样需要传入一个参数,实现的是 SinkFunction 接口。在这个接口中只需要重写一个方法 invoke(),用来将指定的值写入到外部系统中。这个方法在每条数据记录到来时都会调用:

default void invoke(IN value, Context context) throws Exception

当然,SinkFuntion 多数情况下同样并不需要我们自己实现。Flink 官方提供了一部分的框架的 Sink 连接器。如:

我们可以看到,像 Kafka 之类流式系统,Flink 提供了完美对接,source/sink 两端都能连接,可读可写;而对于 Elasticsearch、文件系统(FileSystem)、JDBC 等数据存储系统,则只提供了输出写入的 sink 连接器。

输出文件

输出到文件最简单的输出方式,当然就是写入文件了。对应着读取文件作为输入数据源,Flink 本来也有一些非常简单粗暴的输出到文件的预实现方法:如 writeAsText()、writeAsCsv(),可以直接将输出结果保存到文本文件或 Csv 文件。但我们知道,这种方式是不支持同时写入一份文件的;所以我们往往会将最后的 Sink 操作并行度设为 1,这就大大拖慢了系统效率;而且对于故障恢复后的状态一致性,也没有任何保证。所以目前这些简单的方法已经要被弃用。

Flink 为此专门提供了一个流式文件系统的连接器:StreamingFileSink,它继承自抽象类RichSinkFunction,而且集成了 Flink 的检查点(checkpoint)机制,用来保证精确一次(exactly once)的一致性语义。

StreamingFileSink 为批处理和流处理提供了一个统一的 Sink,它可以将分区文件写入 Flink支持的文件系统。它可以保证精确一次的状态一致性,大大改进了之前流式文件 Sink 的方式。它的主要操作是将数据写入桶(buckets),每个桶中的数据都可以分割成一个个大小有限的分区文件,这样一来就实现真正意义上的分布式文件存储。我们可以通过各种配置来控制“分桶”的操作;默认的分桶方式是基于时间的,我们每小时写入一个新的桶。换句话说,每个桶内保存的文件,记录的都是 1 小时的输出数据。

StreamingFileSink 支持行编码(Row-encoded)和批量编码(Bulk-encoded,比如 Parquet)格式。这两种不同的方式都有各自的构建器(builder),调用方法也非常简单,可以直接调用StreamingFileSink 的静态方法:

  • 行编码:StreamingFileSink.forRowFormat(basePath,rowEncoder)。
  • 批量编码:StreamingFileSink.forBulkFormat(basePath,bulkWriterFactory)。

在创建行或批量编码 Sink 时,我们需要传入两个参数,用来指定存储桶的基本路径(basePath)和数据的编码逻辑(rowEncoder 或 bulkWriterFactory):

public class SinkToFileTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStreamSource<Event> stream = env.fromElements(new Event("Mary", 
                                                                    "./home", 1000L),
                                                          new Event("Bob", "./cart", 2000L),
                                                          new Event("Alice", "./prod?id=100", 3000L),
                                                          new Event("Alice", "./prod?id=200", 3500L),
                                                          new Event("Bob", "./prod?id=2", 2500L),
                                                          new Event("Alice", "./prod?id=300", 3600L),
                                                          new Event("Bob", "./home", 3000L),
                                                          new Event("Bob", "./prod?id=1", 2300L),
                                                          new Event("Bob", "./prod?id=3", 3300L));
        StreamingFileSink<String> fileSink = StreamingFileSink
            .<String>forRowFormat(new Path("./output"),
                                  new SimpleStringEncoder<>("UTF-8"))
            .withRollingPolicy(DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
            .withMaxPartSize(1024 * 1024 * 1024)
            .build())
            .build();
        // 将 Event 转换成 String 写入文件
        stream.map(Event::toString).addSink(fileSink);
        env.execute();
    } 
}

这里我们创建了一个简单的文件 Sink,通过.withRollingPolicy()方法指定了一个“滚动策略”。“滚动”的概念在日志文件的写入中经常遇到:因为文件会有内容持续不断地写入,所以我们应该给一个标准,到什么时候就开启新的文件,将之前的内容归档保存。也就是说,上面的代码设置了在以下 3 种情况下,我们就会滚动分区文件:

  • 至少包含 15 分钟的数据
  • 最近 5 分钟没有收到新的数据
  • 文件大小已达到 1 GB

输出到 Kafka

Kafka 是一个分布式的基于发布/订阅的消息系统,本身处理的也是流式数据,所以跟Flink“天生一对”,经常会作为 Flink 的输入数据源和输出系统。Flink 官方为 Kafka 提供了 Source和 Sink 的连接器,我们可以用它方便地从 Kafka 读写数据。如果仅仅是支持读写,那还说明不了 Kafka 和 Flink 关系的亲密;真正让它们密不可分的是,Flink 与 Kafka 的连接器提供了端到端的精确一次(exactly once)语义保证,这在实际项目中是最高级别的一致性保证。

public class SinkToKafkaTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "hadoop102:9092");
        DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");
        stream
            .addSink(new FlinkKafkaProducer<String>(
                "clicks",
                new SimpleStringSchema(),
                properties
            ));
        env.execute();
    } 
}

这里我们可以看到,addSink 传入的参数是一个 FlinkKafkaProducer。这也很好理解,因为需要向 Kafka 写入数据,自然应该创建一个生产者。FlinkKafkaProducer 继承了抽象类TwoPhaseCommitSinkFunction,这是一个实现了**“两阶段提交”**的 RichSinkFunction。两阶段提交提供了 Flink 向 Kafka 写入数据的事务性保证,能够真正做到精确一次(exactly once)的状态一致性。

输出到 Redis

Redis 是一个开源的内存式的数据存储,提供了像字符串(string)、哈希表(hash)、列表(list)、集合(set)、排序集合(sorted set)、位图(bitmap)、地理索引和流(stream)等一系列常用的数据结构。因为它运行速度快、支持的数据类型丰富,在实际项目中已经成为了架构优化必不可少的一员,一般用作数据库、缓存,也可以作为消息代理。

Flink 没有直接提供官方的 Redis 连接器,不过 Bahir 项目还是担任了合格的辅助角色,为我们提供了 Flink-Redis 的连接工具。但版本升级略显滞后,目前连接器版本为 1.0,支持的Scala 版本最新到 2.11。由于我们的测试不涉及到 Scala 的相关版本变化,所以并不影响使用。在实际项目应用中,应该以匹配的组件版本运行。

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

连接器为我们提供了一个 RedisSink,它继承了抽象类 RichSinkFunction,这就是已经实现好的向 Redis 写入数据的 SinkFunction。我们可以直接将 Event 数据输出到 Redis:

public class SinkToRedisTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建一个到 redis 连接的配置
        FlinkJedisPoolConfig conf = new 
            FlinkJedisPoolConfig.Builder().setHost("hadoop102").build();
        env.addSource(new ClickSource())
            .addSink(new RedisSink<Event>(conf, new MyRedisMapper()));
        env.execute();
    } 
}

public static class MyRedisMapper implements RedisMapper<Event> {
    @Override
    public String getKeyFromData(Event e) {
        return e.user;
    }
    @Override
    public String getValueFromData(Event e) {
        return e.url;
    }
    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.HSET, "clicks");
    } 
}

这里 RedisSink 的构造方法需要传入两个参数:

  • JFlinkJedisConfigBase:Jedis 的连接配置;

  • RedisMapper:Redis 映射类接口,说明怎样将数据转换成可以写入 Redis 的类型。

    在这里我们可以看到,保存到 Redis 时调用的命令是 HSET,所以是保存为哈希表(hash),表名为“clicks”;保存的数据以 user 为 key,以 url 为 value,每来一条数据就会做一次转换。

输出到 Elasticsearch

ElasticSearch 是一个分布式的开源搜索和分析引擎,适用于所有类型的数据。ElasticSearch有着简洁的 REST 风格的 API,以良好的分布式特性、速度和可扩展性而闻名,在大数据领域应用非常广泛。

Flink 为 ElasticSearch 专门提供了官方的 Sink 连接器,Flink 1.13 支持当前最新版本的ElasticSearch。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
public class SinkToEsTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L),
            new Event("Alice", "./prod?id=100", 3000L),
            new Event("Alice", "./prod?id=200", 3500L),
            new Event("Bob", "./prod?id=2", 2500L),
            new Event("Alice", "./prod?id=300", 3600L),
            new Event("Bob", "./home", 3000L),
            new Event("Bob", "./prod?id=1", 2300L),
            new Event("Bob", "./prod?id=3", 3300L));
        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("hadoop102", 9200, "http"));
        // 创建一个 ElasticsearchSinkFunction
        ElasticsearchSinkFunction<Event> elasticsearchSinkFunction = new 
            ElasticsearchSinkFunction<Event>() {
            @Override
            public void process(Event element, RuntimeContext ctx, RequestIndexer 
                                indexer) {
                HashMap<String, String> data = new HashMap<>();
                data.put(element.user, element.url);
                IndexRequest request = Requests.indexRequest()
                    .index("clicks")
                    .type("type") // Es 6 必须定义 type
                    .source(data);
                indexer.add(request);
            }
        };
        stream.addSink(new ElasticsearchSink.Builder<Event>(httpHosts, 
                                                            elasticsearchSinkFunction).build());
        stream.addSink(esBuilder.build());
        env.execute();
    } 
}

与 RedisSink 类 似 , 连 接 器 也 为 我 们 实 现 了 写 入 到 Elasticsearch 的SinkFunction——ElasticsearchSink。区别在于,这个类的构造方法是私有(private)的,我们需要使用 ElasticsearchSink 的 Builder 内部静态类,调用它的 build()方法才能创建出真正的SinkFunction。

而 Builder 的构造方法中又有两个参数:

  • httpHosts:连接到的 Elasticsearch 集群主机列表
  • elasticsearchSinkFunction:这并不是我们所说的 SinkFunction,而是用来说明具体处理逻辑、准备数据向 Elasticsearch 发送请求的函数

具体的操作需要重写中 elasticsearchSinkFunction 中的 process 方法,我们可以将要发送的数据放在一个 HashMap 中,包装成 IndexRequest 向外部发送 HTTP 请求。

输出到 MySQL(JDBC)

关系型数据库有着非常好的结构化数据设计、方便的 SQL 查询,是很多企业中业务数据存储的主要形式。MySQL 就是其中的典型代表。尽管在大数据处理中直接与 MySQL 交互的场景不多,但最终处理的计算结果是要给外部应用消费使用的,而外部应用读取的数据存储往往就是 MySQL。所以我们也需要知道如何将数据输出到 MySQL 这样的传统数据库。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>
public class SinkToMySQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
            new Event("Mary", "./home", 1000L),
            new Event("Bob", "./cart", 2000L),
            new Event("Alice", "./prod?id=100", 3000L),
            new Event("Alice", "./prod?id=200", 3500L),
            new Event("Bob", "./prod?id=2", 2500L),
            new Event("Alice", "./prod?id=300", 3600L),
            new Event("Bob", "./home", 3000L),
            new Event("Bob", "./prod?id=1", 2300L),
            new Event("Bob", "./prod?id=3", 3300L));
        stream.addSink(
            JdbcSink.sink(
                "INSERT INTO clicks (user, url) VALUES (?, ?)",
                (statement, r) -> {
                    statement.setString(1, r.user);
                    statement.setString(2, r.url);
                },
                JdbcExecutionOptions.builder()
                .withBatchSize(1000)
                .withBatchIntervalMs(200)
                .withMaxRetries(5)
                .build(),
                new 
                JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:mysql://localhost:3306/userbehavior")
                // 对于 MySQL 5.7,用"com.mysql.jdbc.Driver"
                .withDriverName("com.mysql.cj.jdbc.Driver")
                .withUsername("username")
                .withPassword("password")
                .build()
            )
        );
        env.execute();
    }
}

自定义 Sink 输出

与 Source 类似,Flink 为我们提供了通用的 SinkFunction 接口和对应的 RichSinkDunction抽象类,只要实现它,通过简单地调用 DataStream 的.addSink()方法就可以自定义写入任何外部存储。之前与外部系统的连接,其实都是连接器帮我们实现了 SinkFunction,现在既然没有现成的,我们就只好自力更生了。例如,Flink 并没有提供 HBase 的连接器,所以需要我们自己写。

在实现 SinkFunction 的时候,需要重写的一个关键方法 invoke(),在这个方法中我们就可以实现将流里的数据发送出去的逻辑。

我们这里使用了 SinkFunction 的富函数版本,因为这里我们又使用到了生命周期的概念,创建 HBase 的连接以及关闭 HBase 的连接需要分别放在 open()方法和 close()方法中。

导入依赖:

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>${hbase.version}</version>
</dependency>
public class SinkCustomtoHBase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
            .fromElements("hello", "world")
            .addSink(
            new RichSinkFunction<String>() {
                public org.apache.hadoop.conf.Configuration configuration; // 管理 Hbase 的配置信息,这里因为 Configuration 的重名问题,将类以完整路径导入
                public Connection connection; // 管理 Hbase 连接
                @Override
                public void open(Configuration parameters) throws Exception {
                    super.open(parameters);
                    configuration = HBaseConfiguration.create();
                    configuration.set("hbase.zookeeper.quorum", 
                                      "hadoop102:2181");
                    connection = 
                        ConnectionFactory.createConnection(configuration);
                }
                @Override
                public void invoke(String value, Context context) throws Exception {
                    Table table = connection.getTable(TableName.valueOf("test")); // 表名为 test
                    Put put = new Put("rowkey".getBytes(StandardCharsets.UTF_8)); // 指定 rowkey
                    put.addColumn("info".getBytes(StandardCharsets.UTF_8) // 指定列名
                                  , value.getBytes(StandardCharsets.UTF_8) // 写入的数据
                                  , "1".getBytes(StandardCharsets.UTF_8)); // 写入的数据
                    table.put(put); // 执行 put 操作
                    table.close(); // 将表关闭
                }
                @Override
                public void close() throws Exception {
                    super.close();
                    connection.close(); // 关闭连接
                }
            }
        );
        env.execute();
    } 
}

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

Spring Cloud Nacos实战(三)- 服务消费者注册和负载均衡,服务注册中心对比

目录Nacos服务消费者注册和负载均衡服务消费者具体配置新建Modulepomyml主启动测试远程调用与Ribbon什么是Ribbon具体使用验证Nacos自带负载均衡Nacos服务注册中心对比提升各种服务注册中心对比CAP模型CP原则&#xff1a;一致性 分区容错性原则AP原则&#xff1a;可用性原则 …

Unity 基于Netcode for gameObjects实现局域网同步

注意事项&#xff1a; 1.需要将unity升级到2021.3及以后的版本的稳定版本&#xff0c;使用不稳定的2022版本测试过&#xff0c;存在打包问题&#xff1a; 效果&#xff1a; 所需Package&#xff1a; 1.Netcode for gameObjects 2.Multiplayer Tools 该package是附带Netcode…

基于Spring Boot的零食商店

文章目录项目介绍主要功能截图&#xff1a;登录后台首页个人信息管理用户管理前台首页购物车部分代码展示设计总结项目获取方式&#x1f345; 作者主页&#xff1a;Java韩立 &#x1f345; 简介&#xff1a;Java领域优质创作者&#x1f3c6;、 简历模板、学习资料、面试题库【关…

[Spring] 难理解的Aop编程 |入门?

作者&#xff1a;狮子也疯狂 专栏&#xff1a;《spring开发》 坚持做好每一步&#xff0c;幸运之神自然会驾凌在你的身上 目录一. &#x1f981; 前言二. &#x1f981; 常见概念2.1 常见术语2.2 AOP入门Ⅰ. &#x1f407; 功能场景Ⅱ. &#x1f407; 实现过程2.3 通知类型Ⅰ.…

使用frp配置内网机器访问

frp简介 frp 是一个开源、简洁易用、高性能的内网穿透和反向代理软件&#xff0c;支持 tcp, udp, http, https等协议。frp 项目官网是 https://github.com/fatedier/frp&#xff0c;软件下载地址为https://github.com/fatedier/frp/releases frp工作原理 服务端运行&#xf…

【GO】k8s 管理系统项目[前端部分--项目初始化]

【GO】k8s 管理系统项目[前端部分–项目初始化] 1. 项目概述 API部分已经完成了,着手开始前端部分.构建一个页面展示后端数据. 前端会使用到以下依赖 vue3框架element-plusxterm命令行模拟器nprogress进度条jsonwebtoken jwt token生成和校验json-editor-vue3/codemirror-e…

一文带你读懂Dockerfile

目录 一、概述 二、DockerFile构建过程解析 &#xff08;一&#xff09;Dockerfile内容基础知识 &#xff08;二&#xff09;Docker执行Dockerfile的大致流程 &#xff08;三&#xff09;总结 三、DockerFile常用保留字指令 四、案例 &#xff08;一&#xff09;自定义…

有了java基础,迅速学完Python并做了一份笔记-全套Python,建议收藏

面向过程Python简介Python和Java的解释方式对比Java&#xff1a;源代码 -> 编译成class -> Jvm解释运行Python&#xff1a;源代码 -> Python解释器解释运行我经常和身边的Java开发者开玩笑说&#xff1a;“Java真变态&#xff0c;别的语言都是要么直接编译要么直接解释…

Kaldi语音识别技术(六) ----- DTW和HMM-GMM

Kaldi语音识别技术(六) ----- DTW和HMM-GMM 文章目录Kaldi语音识别技术(六) ----- DTW和HMM-GMM前言一、语音识别概况二、语音识别基本原理三、DTW&#xff08;动态时间弯折&#xff09;算法四、GMM-HMM前言 前面的内容中我们完成了特征的提取,那么本章节我们主要进行理论部分…

IDEA全家桶式讲解 | IDEA安装、使用、断点调试、Git、插件 (第二篇)

目录 一&#xff1a;JavaEE阶段需要掌握的IDEA技能 1. 配置Tomcat 2. 配置Maven 3. IDEA连接数据库 4. 方便的特殊功能 5. 断点调试&#xff08;重点&#xff09; 6. IDEA中常用Git协同开发&#xff08;重点&#xff09; 7. 常用插件安装 一&#xff1a;JavaEE阶段需要…

Julia 语言环境安装

Julia 语言支持以下系统&#xff1a; LinuxFreeBSDmacOSWindowsAndroid Julia 安装包下载地址为&#xff1a;Download Julia。 Github 源码地址&#xff1a;GitHub - JuliaLang/julia: The Julia Programming Language。 国内镜像地址&#xff1a;Index of /julia-releases/…

Spring Boot框架基础介绍

Spring Boot 是一款基于 Spring 框架的开源应用程序开发工具&#xff0c;它旨在简化 Spring 应用程序的配置和开发过程。Spring Boot 提供了一种简单的方式来创建可独立运行的、生产级别的应用程序&#xff0c;并在需要时进行部署。Spring Boot 在微服务架构和云计算环境下得到…

nodejs基于vue垃圾回收分类网站

目录 1 绪论 1 1.1课题背景 1 1.2课题研究现状 1 1.3初步设计方法与实施方案 2 1.4本文研究内容 2 2 系统开发环境 4 2.1 JAVA简介 4 2.2MyEclipse环境配置 4 2.3 B/S结构简介 4 2.4MySQL数据库 5 2.5 SPRINGBOOT框架 5 3 系统分析 6 3.1系统可行性分析 6 3.1.1经济可行性 6 3.…

【C++修炼之路】18.map和set

每一个不曾起舞的日子都是对生命的辜负 map和setmap和set一.关联式容器二.set2.1 set的介绍2.2 set的使用1.set的模板参数列表2.set的构造3.set的迭代器4.set修改操作5.bound函数三.multiset四.map3.1 map的介绍3.2 map的使用1.map的模板参数说明2.pair的介绍3.map的[]重载五.m…

如何构建微服务架构?

相信很多人对微服务架构都会产生这样一些疑问&#xff0c;例如我要何时使用微服务架构?又如何将应用程序分解为微服务?分解后&#xff0c;要如何去搭建微服务架构?同时&#xff0c;在微服务架构中&#xff0c;因为会涉及到多个组件&#xff0c;那么这些组件又可以使用什么技…

[软件工程导论(第六版)]第9章 面向对象方法学引论(复习笔记)

文章目录9.1 面向对象方法学概述要点9.2 面向对象的概念对象9.3 面向对象建模9.4 对象模型9.5 动态模型9.6 功能模型9.7 3种模型之间的关系9.1 面向对象方法学概述要点 面向对象方法学的出发点和基本原则&#xff0c;是尽可能模拟人类习惯的思维方式&#xff0c;使开发软件的方…

CS144-Lab3

概述 在实验0中&#xff0c;你实现了流控制字节流&#xff08;ByteStream&#xff09;的抽象。 在实验1和2中&#xff0c;你实现了将不可靠数据报中的段转换为传入字节流的工具&#xff1a;StreamReassembler和TCPReceiver。 现在&#xff0c;在实验3中&#xff0c;你将实现…

【STM32笔记】低功耗模式配置及避坑汇总

【STM32笔记】低功耗模式配置及配置汇总 文章总结&#xff1a;&#xff08;后续更新以相关文章为准&#xff09; 【STM32笔记】__WFI()&#xff1b;进入不了休眠的可能原因 【STM32笔记】HAL库低功耗模式配置&#xff08;ADC唤醒无法使用、低功耗模式无法烧录解决方案&#x…

kanban系统wekan安装

看板类开源项目排名第一的wekan项目安装比较友好的leantime 下载windows 版本 wekan 进入官网 https://wekan.github.io/ , 留意最新版依赖的技术栈, 比如 WeKan v6.74 依赖的是 Meteor 2.10.0, Node.js 14.21.2, MongoDB 6.0.4。 点击 Offline Window LAN 链接进入 github wik…

「7」线性代数(期末复习)

&#x1f680;&#x1f680;&#x1f680;大家觉不错的话&#xff0c;就恳求大家点点关注&#xff0c;点点小爱心&#xff0c;指点指点&#x1f680;&#x1f680;&#x1f680; 目录 第五章 相似矩阵及二次型 &4&#xff09;对称阵的对角化 &5二次型及其标准型 …