19、Flink 的Table API 和 SQL 中的自定义函数及示例（3）

Flink 系列文章

1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接

13、Flink 的table api与sql的基本概念、通用api介绍及入门示例
14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性
15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例（1）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例（2）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例（3）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例（4）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例（6）
17、Flink 之Table API: Table API 支持的操作（1）
17、Flink 之Table API: Table API 支持的操作（2）
18、Flink的SQL 支持的操作和语法
19、Flink 的Table API 和 SQL 中的内置函数及示例（1）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（2）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（3）
20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL，可以直接提交 SQL 任务到集群上

22、Flink 的table api与sql之创建表的DDL
24、Flink 的table api与sql之Catalogs（介绍、类型、java api和sql实现ddl、java api和sql操作catalog）-1
24、Flink 的table api与sql之Catalogs（java api操作数据库、表）-2
24、Flink 的table api与sql之Catalogs（java api操作视图）-3
24、Flink 的table api与sql之Catalogs（java api操作分区与函数）-4

26、Flink 的SQL之概览与入门示例
27、Flink 的SQL之SELECT (select、where、distinct、order by、limit、集合操作和去重)介绍及详细示例（1）
27、Flink 的SQL之SELECT (SQL Hints 和 Joins)介绍及详细示例（2）
27、Flink 的SQL之SELECT (窗口函数)介绍及详细示例（3）
27、Flink 的SQL之SELECT (窗口聚合)介绍及详细示例（4）
27、Flink 的SQL之SELECT (Group Aggregation分组聚合、Over Aggregation Over聚合 和 Window Join 窗口关联)介绍及详细示例（5）
27、Flink 的SQL之SELECT (Top-N、Window Top-N 窗口 Top-N 和 Window Deduplication 窗口去重)介绍及详细示例（6）
27、Flink 的SQL之SELECT (Pattern Recognition 模式检测)介绍及详细示例（7）
28、Flink 的SQL之DROP 、ALTER 、INSERT 、ANALYZE 语句
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（1）
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（2）
30、Flink SQL之SQL 客户端（通过kafka和filesystem的例子介绍了配置文件使用-表、视图等）
32、Flink table api和SQL 之用户自定义 Sources & Sinks实现及详细示例
41、Flink之Hive 方言介绍及详细示例
42、Flink 的table api与sql之Hive Catalog
43、Flink之Hive 读写及详细验证示例
44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例–网上有些说法好像是错误的

文章目录

  • Flink 系列文章
    • 5、聚合函数
      • 1）、示例
    • 6、表值聚合函数
      • 1）、示例1- 计算topN
      • 2）、示例2 - emitUpdateWithRetract 方法使用（老版本Planner可用）

本文介绍了标量聚合函数和表值聚合函数的自定义实现，分别以具体的示例进行展示。特别需要提醒的是表值聚合函数自定义实现时针对emitValue和emitUpdateWithRetract方法的不同版本实现要求，该处在其官网上没有特别的说明，会导致运行异常，具体原因及解决办法在示例2emitUpdateWithRetract中有说明。
本文依赖flink集群能正常使用。
本文分为2个部分，即标量聚合函数以及表值聚合函数的自定义实现。
本文的示例如无特殊说明则是在Flink 1.17版本中运行。

5、聚合函数

自定义聚合函数（UDAGG）是把一个表（一行或者多行，每行可以有一列或者多列）聚合成一个标量值。
在这里插入图片描述
上面的图片展示了一个聚合的例子。
假设有一个关于饮料的表。表里面有三个字段，分别是 id、name、price，表里有 5 行数据。
假设需要找到所有饮料里最贵的饮料的价格，即执行一个 max() 聚合。
需要遍历所有 5 行数据，而结果就只有一个数值。

自定义聚合函数是通过扩展 AggregateFunction 来实现的。AggregateFunction 的工作过程如下。
首先，它需要一个 accumulator，它是一个数据结构，存储了聚合的中间结果。通过调用 AggregateFunction 的 createAccumulator() 方法创建一个空的 accumulator。
接下来，对于每一行数据，会调用 accumulate() 方法来更新 accumulator。
当所有的数据都处理完了之后，通过调用 getValue 方法来计算和返回最终的结果。

下面几个方法是每个 AggregateFunction 必须要实现的：

  • createAccumulator()
  • accumulate()
  • getValue()

Flink 的类型推导在遇到复杂类型的时候可能会推导出错误的结果，比如那些非基本类型和普通的 POJO 类型的复杂类型。所以跟 ScalarFunction 和 TableFunction 一样，AggregateFunction 也提供了 AggregateFunction#getResultType() 和 AggregateFunction#getAccumulatorType() 来分别指定返回值类型和 accumulator 的类型，两个函数的返回值类型也都是 TypeInformation。

除了上面的方法，还有几个方法可以选择实现。这些方法有些可以让查询更加高效，而有些是在某些特定场景下必须要实现的。
例如，如果聚合函数用在会话窗口（当两个会话窗口合并的时候需要 merge 他们的 accumulator）的话，merge() 方法就是必须要实现的。

AggregateFunction 的以下方法在某些场景下是必须实现的：

  • retract() 在 bounded OVER 窗口中是必须实现的。
  • merge() 在许多批式聚合和会话以及滚动窗口聚合中是必须实现的。除此之外，这个方法对于优化也很多帮助。例如，两阶段聚合优化就需要所有的 AggregateFunction 都实现 merge 方法。
  • resetAccumulator() 在许多批式聚合中是必须实现的。

AggregateFunction 的所有方法都必须是 public 的，不能是 static 的，而且名字必须跟上面写的一样。createAccumulator、getValue、getResultType 以及 getAccumulatorType 这几个函数是在抽象类 AggregateFunction 中定义的，而其他函数都是约定的方法。如果要定义一个聚合函数，你需要扩展 org.apache.flink.table.functions.AggregateFunction，并且实现一个（或者多个）accumulate 方法。accumulate 方法可以重载，每个方法的参数类型不同，并且支持变长参数。

AggregateFunction 的所有方法的详细文档说明如下。


/*** Base class for user-defined aggregates and table aggregates.* 用户定义聚合和表聚合的基类。** @param <T>   the type of the aggregation result.* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the*             aggregated values which are needed to compute an aggregation result.* 聚合累加器的类型。累加器用于保存计算聚合结果所需的聚合值。*/
public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {/*** Creates and init the Accumulator for this (table)aggregate function.* 创建和初始化aggregate function 的Accumulator 方法** @return the accumulator with the initial value*/public ACC createAccumulator(); // MANDATORY/*** Returns the TypeInformation of the (table)aggregate function's result.* 返回aggregate function的结果类型TypeInformation ** @return The TypeInformation of the (table)aggregate function's result or null if the result*         type should be automatically inferred.* 返回aggregate function的结果类型TypeInformation ，如果结果为null，则会自动推导类型*/public TypeInformation<T> getResultType = null; // PRE-DEFINED/*** Returns the TypeInformation of the (table)aggregate function's accumulator.* 返回aggregate function's accumulator 的类型TypeInformation ** @return The TypeInformation of the (table)aggregate function's accumulator or null if the*         accumulator type should be automatically inferred.*/public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
}/*** Base class for aggregation functions.** @param <T>   the type of the aggregation result* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the*             aggregated values which are needed to compute an aggregation result.*             AggregateFunction represents its state using accumulator, thereby the state of the*             AggregateFunction must be put into the accumulator.* acc aggregation accumulator的类型，accumulator 用来保存计算聚合结果所需的聚合值。* AggregateFunction使用累加器表示其状态，从而表示AggregateFunction必须放入累加器*/
public abstract class AggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {/** Processes the input values and update the provided accumulator instance. The method* accumulate can be overloaded with different custom types and arguments. An AggregateFunction* requires at least one accumulate() method.* 处理输入值并更新提供的累加器实例。方法accumulate 可以用不同的自定义类型和参数重载。* 聚合函数至少需要一个accumulate（）方法。** @param accumulator  the accumulator which contains the current aggregated results* @param [user defined inputs] the input value (usually obtained from a new arrived data).*/public void accumulate(ACC accumulator, [user defined inputs]); // MANDATORY/*** Retracts the input values from the accumulator instance. The current design assumes the* inputs are the values that have been previously accumulated. The method retract can be* overloaded with different custom types and arguments. This function must be implemented for* datastream bounded over aggregate.* 收回累加器实例中的输入值。当前设计假设输入是先前累积的值。收回方法可以是重载了不同的自定义类型和参数。* 此功能在datastream的有界流基于over aggregate必须被实现。** @param accumulator the accumulator which contains the current aggregated results* @param [user defined inputs] the input value (usually obtained from a new arrived data).*/public void retract(ACC accumulator, [user defined inputs]); // OPTIONAL/*** Merges a group of accumulator instances into one accumulator instance. This function must be* implemented for datastream session window grouping aggregate and bounded grouping aggregate.* 将一组accumulator 实例合并为一个accumulator 实例。* 该函数在datastream session window的分组聚合 和 有界流的分组聚合必须实现。* @param accumulator  the accumulator which will keep the merged aggregate results. It should*                     be noted that the accumulator may contain the previous aggregated*                     results. Therefore user should not replace or clean this instance in the*                     custom merge method. 累加器，用于保存合并后的聚合结果。* 					  应该注意的是，累加器可以包含先前的聚合结果。* 					  因此，用户不应在自定义合并方法中替换或清除此实例。* @param its          an {@link java.lang.Iterable} pointed to a group of accumulators that will be*                     merged.*/public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL/*** Called every time when an aggregation result should be materialized.* The returned value could be either an early and incomplete result* (periodically emitted as data arrive) or the final result of the aggregation.* 每次应该具体化（materialized）聚合结果时调用。* 返回的值可能是早期且不完整的结果（随着数据的到达而定期发出），也可能是聚合的最终结果。** @param accumulator the accumulator which contains the current aggregated results* @return the aggregation result*/public T getValue(ACC accumulator); // MANDATORY/*** Resets the accumulator for this [[AggregateFunction]]. This function must be implemented for* bounded grouping aggregate.* 重置此[[AggregateFunction]]的累加器。必须为有界分组聚合实现此函数。** @param accumulator  the accumulator which needs to be reset*/public void resetAccumulator(ACC accumulator); // OPTIONAL/*** Returns true if this AggregateFunction can only be applied in an OVER window.* 如果此AggregateFunction只能在OVER窗口中应用，则返回true。** @return true if the AggregateFunction requires an OVER window, false otherwise.*/public Boolean requiresOver = false; // PRE-DEFINED
}

1）、示例

该示例包含以下三个功能：

  • 定义一个聚合函数来计算某一列的加权平均
  • 在 TableEnvironment 中注册函数
  • 在查询中使用函数

为了计算加权平均值，accumulator 需要存储加权总和以及数据的条数。
在例子里，定义了一个类 Aalan_WeightedAvgAccum 来作为 accumulator。Flink 的 checkpoint 机制会自动保存 accumulator，在失败时进行恢复，以此来保证精确一次的语义。

例子的WeightedAvgAggregateFunction（自定义聚合函数）的 accumulate 方法有三个输入参数。
第一个是 Aalan_WeightedAvgAccum accumulator
另外两个是用户自定义的输入：输入的值 ivalue（balance） 和 输入的权重 iweight（age）。

尽管 retract()、merge()、resetAccumulator() 这几个方法在大多数聚合类型中都不是必须实现的，样例中提供了他们的实现。

在 Scala 样例中也是用的是 Java 的基础类型，并且定义了 getResultType() 和 getAccumulatorType()，因为 Flink 的类型推导对于 Scala 的类型推导做的不是很好。

import static org.apache.flink.table.api.Expressions.$;import java.util.Arrays;
import java.util.Iterator;
import java.util.List;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** @author alanchan**/
public class TestUDAGGDemo {// 加权平均累加器bean，加上名称，以示区别，避免混淆public static class Aalan_WeightedAvgAccum {public long sum = 0;public int count = 0;}// 聚合函数的自定义实现，计算加权平均public static class WeightedAvgAggregateFunction extends AggregateFunction<Long, Aalan_WeightedAvgAccum> {/*** 创建和初始化aggregate function 的Accumulator 方法*/@Overridepublic Aalan_WeightedAvgAccum createAccumulator() {return new Aalan_WeightedAvgAccum();}/*** 每次应该具体化（materialized）聚合结果时调用。 返回的值可能是早期且不完整的结果（随着数据的到达而定期发出），也可能是聚合的最终结果。*/@Overridepublic Long getValue(Aalan_WeightedAvgAccum acc) {if (acc.count == 0) {return null;} else {return acc.sum / acc.count;}}/*** 处理输入值并更新提供的累加器实例。方法accumulate 可以用不同的自定义类型和参数重载。聚合函数至少需要一个accumulate（）方法。* * @param acc     累加器bean,包含当前汇总结果的累加器* @param iValue  输入的需要的累加值* @param iWeight 输入的需要累加的值的权重*/public void accumulate(Aalan_WeightedAvgAccum acc, long iValue, int iWeight) {acc.sum += iValue * iWeight;acc.count += iWeight;}/*** 收回累加器实例中的输入值。当前设计假设输入是先前累积的值。收回方法可以是重载了不同的自定义类型和参数。 此功能在datastream的有界流基于over* aggregate必须被实现。* * @param acc     累加器bean,包含当前汇总结果的累加器* @param iValue  输入的需要的累加值* @param iWeight 输入的需要累加的值的权重*/public void retract(Aalan_WeightedAvgAccum acc, long iValue, int iWeight) {acc.sum -= iValue * iWeight;acc.count -= iWeight;}/*** 将一组accumulator 实例合并为一个accumulator 实例。 该函数在datastream session window的分组聚合 和* 有界流的分组聚合必须实现。* * @param acc 累加器，用于保存合并后的聚合结果。 应该注意的是，累加器可以包含先前的聚合结果。 因此，用户不应在自定义合并方法中替换或清除此实例。* @param it  指向将被合并的一组累加器的Iterable。*/public void merge(Aalan_WeightedAvgAccum acc, Iterable<Aalan_WeightedAvgAccum> it) {Iterator<Aalan_WeightedAvgAccum> iter = it.iterator();while (iter.hasNext()) {Aalan_WeightedAvgAccum a = iter.next();acc.count += a.count;acc.sum += a.sum;}}/*** 重置此[[AggregateFunction]]的累加器。必须为有界分组聚合实现此函数。* * @param acc*/public void resetAccumulator(Aalan_WeightedAvgAccum acc) {acc.count = 0;acc.sum = 0L;}}@Data@NoArgsConstructor@AllArgsConstructorpublic static class User {private long id;private String name;private int age;private long balance;private Long rowtime;}final static List<User> userList = Arrays.asList(new User(1L, "alan", 18, 20, 1698742358391L), new User(2L, "alan", 19, 25, 1698742359396L),new User(3L, "alan", 25, 30, 1698742360407L), new User(4L, "alanchan", 28, 35, 1698742361409L), new User(5L, "alanchan", 29, 35, 1698742362424L));public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv = StreamTableEnvironment.create(env);// 将聚合函数注册为函数tenv.registerFunction("alan_weightavgAF", new WeightedAvgAggregateFunction());DataStream<User> users = env.fromCollection(userList);Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("age"), $("balance"), $("rowtime"));tenv.createTemporaryView("user_view", users);// 使用函数String sql = "SELECT name, alan_weightavgAF(balance, age) AS avgPoints FROM user_view GROUP BY name";Table result = tenv.sqlQuery(sql);DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);resultDS.print();
//		16> (true,+I[alanchan, 35])
//		2> (true,+I[alan, 20])
//		2> (false,-U[alan, 20])
//		2> (true,+U[alan, 22])
//		2> (false,-U[alan, 22])
//		2> (true,+U[alan, 25])env.execute();}}

6、表值聚合函数

自定义表值聚合函数（UDTAGG）可以把一个表（一行或者多行，每行有一列或者多列）聚合成另一张表，结果中可以有多行多列。
在这里插入图片描述
上图展示了一个表值聚合函数的例子。
假设有一个饮料的表，这个表有 3 列，分别是 id、name 和 price，一共有 5 行。
假设需要找到价格最高的两个饮料，类似于 top2() 表值聚合函数。
需要遍历所有 5 行数据，结果是有 2 行数据的一个表。

用户自定义表值聚合函数是通过扩展 TableAggregateFunction 类来实现的。
一个 TableAggregateFunction 的工作过程如下。
首先，它需要一个 accumulator，这个 accumulator 负责存储聚合的中间结果。 通过调用 TableAggregateFunction 的 createAccumulator 方法来构造一个空的 accumulator。
接下来，对于每一行数据，会调用 accumulate 方法来更新 accumulator。
当所有数据都处理完之后，调用 emitValue 方法来计算和返回最终的结果。

下面几个 TableAggregateFunction 的方法是必须要实现的：

  • createAccumulator()
  • accumulate()

Flink 的类型推导在遇到复杂类型的时候可能会推导出错误的结果，比如那些非基本类型和普通的 POJO 类型的复杂类型。所以类似于 ScalarFunction 和 TableFunction，TableAggregateFunction 也提供了 TableAggregateFunction#getResultType() 和 TableAggregateFunction#getAccumulatorType() 方法来指定返回值类型和 accumulator 的类型，这两个方法都需要返回 TypeInformation。

除了上面的方法，还有几个其他的方法可以选择性的实现。有些方法可以让查询更加高效，而有些方法对于某些特定场景是必须要实现的。比如，在会话窗口（当两个会话窗口合并时会合并两个 accumulator）中使用聚合函数时，必须要实现merge() 方法。

下面几个 TableAggregateFunction 的方法在某些特定场景下是必须要实现的：

  • retract() 在 bounded OVER 窗口中的聚合函数必须要实现。
  • merge() 在许多批式聚合和以及流式会话和滑动窗口聚合中是必须要实现的。
  • resetAccumulator() 在许多批式聚合中是必须要实现的。
  • emitValue() 在批式聚合以及窗口聚合中是必须要实现的。

下面的 TableAggregateFunction 的方法可以提升流式任务的效率：

  • emitUpdateWithRetract() 在 retract 模式下，该方法负责发送被更新的值。

emitValue 方法会发送所有 accumulator 给出的结果。拿 TopN 来说，emitValue 每次都会发送所有的最大的 n 个值。这在流式任务中可能会有一些性能问题。为了提升性能，用户可以实现 emitUpdateWithRetract 方法。这个方法在 retract 模式下会增量的输出结果，比如有数据更新了，我们必须要撤回老的数据，然后再发送新的数据。如果定义了 emitUpdateWithRetract 方法，那它会优先于 emitValue 方法被使用，因为一般认为 emitUpdateWithRetract 会更加高效，因为它的输出是增量的。

TableAggregateFunction 的所有方法都必须是 public 的、非 static 的，而且名字必须跟上面提到的一样。createAccumulator、getResultType 和 getAccumulatorType 这三个方法是在抽象父类 TableAggregateFunction 中定义的，而其他的方法都是约定的方法。要实现一个表值聚合函数，你必须扩展 org.apache.flink.table.functions.TableAggregateFunction，并且实现一个（或者多个）accumulate 方法。accumulate 方法可以有多个重载的方法，也可以支持变长参数。

TableAggregateFunction 的所有方法的详细文档说明如下，其中部分与AggregateFunction 类似的方法不再赘述。

/*** Base class for user-defined aggregates and table aggregates.** @param <T>   the type of the aggregation result.* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the*             aggregated values which are needed to compute an aggregation result.*/
public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {/*** Creates and init the Accumulator for this (table)aggregate function.** @return the accumulator with the initial value*/public ACC createAccumulator(); // MANDATORY/*** Returns the TypeInformation of the (table)aggregate function's result.** @return The TypeInformation of the (table)aggregate function's result or null if the result*         type should be automatically inferred.*/public TypeInformation<T> getResultType = null; // PRE-DEFINED/*** Returns the TypeInformation of the (table)aggregate function's accumulator.** @return The TypeInformation of the (table)aggregate function's accumulator or null if the*         accumulator type should be automatically inferred.*/public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
}/*** Base class for table aggregation functions.** @param <T>   the type of the aggregation result* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the*             aggregated values which are needed to compute a table aggregation result.*             TableAggregateFunction represents its state using accumulator, thereby the state of*             the TableAggregateFunction must be put into the accumulator.*/
public abstract class TableAggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {/** Processes the input values and update the provided accumulator instance. The method* accumulate can be overloaded with different custom types and arguments. A TableAggregateFunction* requires at least one accumulate() method.** @param accumulator           the accumulator which contains the current aggregated results* @param [user defined inputs] the input value (usually obtained from a new arrived data).*/public void accumulate(ACC accumulator, [user defined inputs]); // MANDATORY/*** Retracts the input values from the accumulator instance. The current design assumes the* inputs are the values that have been previously accumulated. The method retract can be* overloaded with different custom types and arguments. This function must be implemented for* datastream bounded over aggregate.** @param accumulator           the accumulator which contains the current aggregated results* @param [user defined inputs] the input value (usually obtained from a new arrived data).*/public void retract(ACC accumulator, [user defined inputs]); // OPTIONAL/*** Merges a group of accumulator instances into one accumulator instance. This function must be* implemented for datastream session window grouping aggregate and bounded grouping aggregate.** @param accumulator  the accumulator which will keep the merged aggregate results. It should*                     be noted that the accumulator may contain the previous aggregated*                     results. Therefore user should not replace or clean this instance in the*                     custom merge method.* @param its          an {@link java.lang.Iterable} pointed to a group of accumulators that will be*                     merged.*/public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL/*** Called every time when an aggregation result should be materialized. The returned value* could be either an early and incomplete result  (periodically emitted as data arrive) or* the final result of the  aggregation.* 每次应该具体化聚合结果时调用。返回的值可能是早期且不完整的结果（随着数据的到达而定期发出），也可能是聚合的最终结果。** @param accumulator the accumulator which contains the current*                    aggregated results* @param out         the collector used to output data*/public void emitValue(ACC accumulator, Collector<T> out); // OPTIONAL/*** Called every time when an aggregation result should be materialized. The returned value* could be either an early and incomplete result (periodically emitted as data arrive) or* the final result of the aggregation.** Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated.* This method outputs data incrementally in retract mode, i.e., once there is an update, we* have to retract old records before sending new updated ones. The emitUpdateWithRetract* method will be used in preference to the emitValue method if both methods are defined in the* table aggregate function, because the method is treated to be more efficient than emitValue* as it can outputvalues incrementally.* 每次应该具体化聚合结果时调用。返回的值可能是早期且不完整的结果（随着数据的到达而定期发出），也可能是聚合的最终结果。* 与emitValue不同，emitUpdateWithRetract用于发出已更新的值。* 该方法以收回模式递增地输出数据，在发送新的更新记录之前，我们必须收回旧记录。* 如果表聚合函数中定义了两个方法，则emitUpdateWithRetract方法将优先于emitValue方法，* 因为该方法被认为比emitValue更有效，因为它可以增量输出值。* * @param accumulator the accumulator which contains the current*                    aggregated results* @param out         the retractable collector used to output data. Use collect method*                    to output(add) records and use retract method to retract(delete)*                    records.*/public void emitUpdateWithRetract(ACC accumulator, RetractableCollector<T> out); // OPTIONAL/*** Collects a record and forwards it. The collector can output retract messages with the retract* method. Note: only use it in {@code emitRetractValueIncrementally}.*/public interface RetractableCollector<T> extends Collector<T> {/*** Retract a record.** @param record The record to retract.*/void retract(T record);}
}

1）、示例1- 计算topN

下面的例子展示了如何

  • 定义一个 TableAggregateFunction 来计算给定列的最大的 3 个值
  • 在 TableEnvironment 中注册函数
  • 在 Table API 查询中使用函数（当前只在 Table API 中支持 TableAggregateFunction）

为了计算最大的 3 个值，accumulator 需要保存当前看到的最大的 3 个值。
在例子中，定义了类 TopAccum 来作为 accumulator。Flink 的 checkpoint 机制会自动保存 accumulator，并且在失败时进行恢复，来保证精确一次的语义。

TopTableAggregateFunction 表值聚合函数（TableAggregateFunction）的 accumulate() 方法有两个输入，
第一个是 TopAccum accumulator，
另一个是用户定义的输入：输入的值 v。

尽管 merge() 方法在大多数聚合类型中不是必须的，也在样例中提供了它的实现。

在 Scala 样例中也使用的是 Java 的基础类型，并且定义了 getResultType() 和 getAccumulatorType() 方法，因为 Flink 的类型推导对于 Scala 的类型推导支持的不是很好。

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;
import java.util.List;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** @author alanchan**/
public class TestUDTAGGDemo {/*** Accumulator for Top3**/@Datapublic static class TopAccum {public Integer first;public Integer second;public Integer third;}public static class TopTableAggregateFunction extends TableAggregateFunction<Tuple2<Integer, Integer>, TopAccum> {@Overridepublic TopAccum createAccumulator() {TopAccum acc = new TopAccum();acc.first = Integer.MIN_VALUE;acc.second = Integer.MIN_VALUE;acc.third = Integer.MIN_VALUE;return acc;}public void accumulate(TopAccum acc, Integer v) {if (v > acc.first) {acc.third = acc.second;acc.second = acc.first;acc.first = v;} else if (v > acc.second) {acc.third = acc.second;acc.second = v;} else if (v > acc.third) {acc.third = v;}}public void merge(TopAccum acc, java.lang.Iterable<TopAccum> iterable) {for (TopAccum otherAcc : iterable) {accumulate(acc, otherAcc.first);accumulate(acc, otherAcc.second);accumulate(acc, otherAcc.third);}}public void emitValue(TopAccum acc, Collector<Tuple2<Integer, Integer>> out) {
//			System.out.println("acc:"+acc);// emit the value and rankif (acc.first != Integer.MIN_VALUE) {out.collect(Tuple2.of(acc.first, 1));}if (acc.second != Integer.MIN_VALUE) {out.collect(Tuple2.of(acc.second, 2));}if (acc.third != Integer.MIN_VALUE) {out.collect(Tuple2.of(acc.third, 3));}}}@Data@NoArgsConstructor@AllArgsConstructorpublic static class User {private long id;private String name;private int age;private int balance;private Long rowtime;}final static List<User> userList = Arrays.asList(new User(1L, "alan", 18, 20, 1698742358391L), new User(2L, "alan", 19, 25, 1698742359396L),new User(3L, "alan", 25, 30, 1698742360407L), new User(4L, "alanchan", 28, 35, 1698742361409L), new User(5L, "alanchan", 29, 35, 1698742362424L));public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv = StreamTableEnvironment.create(env);// 将聚合函数注册为函数tenv.registerFunction("top", new TopTableAggregateFunction());DataStream<User> users = env.fromCollection(userList);Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("age"), $("balance"), $("rowtime"));// 使用函数Table result = usersTable.groupBy($("name")).flatAggregate(call("top", $("balance"))).select($("name"), $("f0").as("balance"), $("f1").as("rank"));DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);resultDS.print();
//		2> (true,+I[alan, 20, 1])
//		16> (true,+I[alanchan, 35, 1])
//		2> (false,-D[alan, 20, 1])
//		16> (false,-D[alanchan, 35, 1])
//		2> (true,+I[alan, 25, 1])
//		16> (true,+I[alanchan, 35, 1])
//		2> (true,+I[alan, 20, 2])
//		16> (true,+I[alanchan, 35, 2])
//		2> (false,-D[alan, 25, 1])
//		2> (false,-D[alan, 20, 2])
//		2> (true,+I[alan, 30, 1])
//		2> (true,+I[alan, 25, 2])
//		2> (true,+I[alan, 20, 3])env.execute();}}

2）、示例2 - emitUpdateWithRetract 方法使用（老版本Planner可用）

下面的例子展示了如何使用 emitUpdateWithRetract 方法来只发送更新的数据。
为了只发送更新的结果，accumulator 保存了上一次的最大的3个值，也保存了当前最大的3个值。

如果 TopN 中的 n 非常大，这种既保存上次的结果，也保存当前的结果的方式不太高效。

一种解决这种问题的方式是把输入数据直接存储到 accumulator 中，然后在调用 emitUpdateWithRetract 方法时再进行计算。

需要特别说明的是

下面的示例需要使用到useOldPlanner，对应的planner的maven依赖见下文

<!-- flink执行计划,这是1.9版本之前的--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.11</artifactId><version>1.13.6</version></dependency>

如果flink的版本比较高的话，下面的示例将不能运行，因为新版本的Builder没有useOldPlanner()方法了，已经移除。不能构造EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();

		//新版本该方法已经被移除@Deprecatedpublic Builder useOldPlanner() {this.plannerClass = OLD_PLANNER_FACTORY;this.executorClass = OLD_EXECUTOR_FACTORY;return this;}

如果使用OldPlanner的话，emitValue和emitUpdateWithRetract仅需定义一个就可以了，并且emitUpdateWithRetract的优先级大于emitValue。但是在Blink Planner里，只看有没有定义emitValue。

也即在Blink Planner中，只能使用emitValue，不能使用emitUpdateWithRetract。
否则会报如下异常
Exception in thread “main” org.apache.flink.table.api.ValidationException: Could not find an implementation method ‘emitValue’ in class ‘org.tablesql.udf.TestUDTAGGDemo2$TopNTableAggregateFunction’ for function ‘TopNTableAggregateFunction’ that matches the following signature:

void emitValue(org.tablesql.udf.TestUDTAGGDemo2.TopNAccum, org.apache.flink.util.Collector)

具体示例如下


import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;
import java.util.List;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** @author alanchan**/
public class TestUDTAGGDemo2 {@Datapublic static class TopNAccum {public Integer first;public Integer second;public Integer third;public Integer oldFirst;public Integer oldSecond;public Integer oldThird;}/*** 自定义聚合函数实现* * @author alanchan**/public static class TopNTableAggregateFunction extends TableAggregateFunction<Tuple2<Integer, Integer>, TopNAccum> {@Overridepublic TopNAccum createAccumulator() {TopNAccum topNAccum = new TopNAccum();topNAccum.first = Integer.MIN_VALUE;topNAccum.second = Integer.MIN_VALUE;topNAccum.third = Integer.MIN_VALUE;topNAccum.oldFirst = Integer.MIN_VALUE;topNAccum.oldSecond = Integer.MIN_VALUE;topNAccum.oldThird = Integer.MIN_VALUE;return topNAccum;}public void accumulate(TopNAccum acc, Integer v) {if (v > acc.first) {acc.third = acc.second;acc.second = acc.first;acc.first = v;} else if (v > acc.second) {acc.third = acc.second;acc.second = v;} else if (v > acc.third) {acc.third = v;}}public void emitUpdateWithRetract(TopNAccum acc, RetractableCollector<Tuple2<Integer, Integer>> out) {System.out.println("emitUpdateWithRetract----acc:" + acc);if (!acc.first.equals(acc.oldFirst)) {// if there is an update, retract old value then emit new value.if (acc.oldFirst != Integer.MIN_VALUE) {out.retract(Tuple2.of(acc.oldFirst, 1));}out.collect(Tuple2.of(acc.first, 1));acc.oldFirst = acc.first;}if (!acc.second.equals(acc.oldSecond)) {// if there is an update, retract old value then emit new value.if (acc.oldSecond != Integer.MIN_VALUE) {out.retract(Tuple2.of(acc.oldSecond, 2));}out.collect(Tuple2.of(acc.second, 2));acc.oldSecond = acc.second;}if (!acc.third.equals(acc.oldThird)) {// if there is an update, retract old value then emit new value.if (acc.oldThird != Integer.MIN_VALUE) {out.retract(Tuple2.of(acc.oldThird, 3));}out.collect(Tuple2.of(acc.third, 3));acc.oldThird = acc.third;}}}@Data@NoArgsConstructor@AllArgsConstructorpublic static class User {private long id;private String name;private int age;private int balance;private Long rowtime;}final static List<User> userList = Arrays.asList(new User(1L, "alan", 18, 20, 1698742358391L), new User(2L, "alan", 19, 25, 1698742359396L),new User(3L, "alan", 25, 30, 1698742360407L), new User(11L, "alan", 28, 31, 1698742358391L), new User(12L, "alan", 29, 32, 1698742359396L),new User(13L, "alan", 35, 35, 1698742360407L), new User(23L, "alan", 45, 36, 1698742360407L), new User(14L, "alanchan", 28, 15, 1698742361409L), new User(15L, "alanchan", 29, 16, 1698742362424L), new User(24L, "alanchan", 30, 20, 1698742361409L),new User(25L, "alanchan", 31, 22, 1698742362424L), new User(34L, "alanchan", 32, 24, 1698742361409L), new User(35L, "alanchan", 33, 26, 1698742362424L),new User(44L, "alanchan", 34, 28, 1698742361409L), new User(55L, "alanchan", 35, 35, 1698742362424L));public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.setParallelism(1);EnvironmentSettings settings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();StreamTableEnvironment tenv = StreamTableEnvironment.create(env, settings);// 将聚合函数注册为函数tenv.registerFunction("topN", new TopNTableAggregateFunction());DataStream<User> users = env.fromCollection(userList);Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("age"), $("balance"), $("rowtime"));// 使用函数Table result = usersTable.groupBy($("name")).flatAggregate(call("topN", $("balance"))).select($("name"), $("f0").as("balance"), $("f1").as("rank"));DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);resultDS.print();env.execute();}}

运行结果如下：

emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=20, second=-2147483648, third=-2147483648, oldFirst=-2147483648, oldSecond=-2147483648, oldThird=-2147483648)
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=15, second=-2147483648, third=-2147483648, oldFirst=-2147483648, oldSecond=-2147483648, oldThird=-2147483648)
14> (true,+I[alan, 20, 1])
9> (true,+I[alanchan, 15, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=25, second=20, third=-2147483648, oldFirst=20, oldSecond=-2147483648, oldThird=-2147483648)
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=16, second=15, third=-2147483648, oldFirst=15, oldSecond=-2147483648, oldThird=-2147483648)
14> (false,+I[alan, 20, 1])
14> (true,+I[alan, 25, 1])
14> (true,+I[alan, 20, 2])
9> (false,+I[alanchan, 15, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=30, second=25, third=20, oldFirst=25, oldSecond=20, oldThird=-2147483648)
14> (false,+I[alan, 25, 1])
9> (true,+I[alanchan, 16, 1])
14> (true,+I[alan, 30, 1])
14> (false,+I[alan, 20, 2])
9> (true,+I[alanchan, 15, 2])
14> (true,+I[alan, 25, 2])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=20, second=16, third=15, oldFirst=16, oldSecond=15, oldThird=-2147483648)
14> (true,+I[alan, 20, 3])
9> (false,+I[alanchan, 16, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=31, second=30, third=25, oldFirst=30, oldSecond=25, oldThird=20)
9> (true,+I[alanchan, 20, 1])
9> (false,+I[alanchan, 15, 2])
14> (false,+I[alan, 30, 1])
9> (true,+I[alanchan, 16, 2])
14> (true,+I[alan, 31, 1])
9> (true,+I[alanchan, 15, 3])
14> (false,+I[alan, 25, 2])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=22, second=20, third=16, oldFirst=20, oldSecond=16, oldThird=15)
14> (true,+I[alan, 30, 2])
9> (false,+I[alanchan, 20, 1])
14> (false,+I[alan, 20, 3])
9> (true,+I[alanchan, 22, 1])
14> (true,+I[alan, 25, 3])
9> (false,+I[alanchan, 16, 2])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=32, second=31, third=30, oldFirst=31, oldSecond=30, oldThird=25)
9> (true,+I[alanchan, 20, 2])
9> (false,+I[alanchan, 15, 3])
9> (true,+I[alanchan, 16, 3])
14> (false,+I[alan, 31, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=24, second=22, third=20, oldFirst=22, oldSecond=20, oldThird=16)
14> (true,+I[alan, 32, 1])
9> (false,+I[alanchan, 22, 1])
14> (false,+I[alan, 30, 2])
9> (true,+I[alanchan, 24, 1])
9> (false,+I[alanchan, 20, 2])
14> (true,+I[alan, 31, 2])
9> (true,+I[alanchan, 22, 2])
14> (false,+I[alan, 25, 3])
9> (false,+I[alanchan, 16, 3])
14> (true,+I[alan, 30, 3])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=35, second=32, third=31, oldFirst=32, oldSecond=31, oldThird=30)
9> (true,+I[alanchan, 20, 3])
14> (false,+I[alan, 32, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=26, second=24, third=22, oldFirst=24, oldSecond=22, oldThird=20)
14> (true,+I[alan, 35, 1])
9> (false,+I[alanchan, 24, 1])
14> (false,+I[alan, 31, 2])
9> (true,+I[alanchan, 26, 1])
14> (true,+I[alan, 32, 2])
9> (false,+I[alanchan, 22, 2])
14> (false,+I[alan, 30, 3])
9> (true,+I[alanchan, 24, 2])
14> (true,+I[alan, 31, 3])
9> (false,+I[alanchan, 20, 3])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=36, second=35, third=32, oldFirst=35, oldSecond=32, oldThird=31)
9> (true,+I[alanchan, 22, 3])
14> (false,+I[alan, 35, 1])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=28, second=26, third=24, oldFirst=26, oldSecond=24, oldThird=22)
14> (true,+I[alan, 36, 1])
9> (false,+I[alanchan, 26, 1])
14> (false,+I[alan, 32, 2])
9> (true,+I[alanchan, 28, 1])
14> (true,+I[alan, 35, 2])
9> (false,+I[alanchan, 24, 2])
9> (true,+I[alanchan, 26, 2])
14> (false,+I[alan, 31, 3])
9> (false,+I[alanchan, 22, 3])
14> (true,+I[alan, 32, 3])
9> (true,+I[alanchan, 24, 3])
emitUpdateWithRetract----acc:TestUDTAGGDemo2.TopNAccum(first=35, second=28, third=26, oldFirst=28, oldSecond=26, oldThird=24)
9> (false,+I[alanchan, 28, 1])
9> (true,+I[alanchan, 35, 1])
9> (false,+I[alanchan, 26, 2])
9> (true,+I[alanchan, 28, 2])
9> (false,+I[alanchan, 24, 3])
9> (true,+I[alanchan, 26, 3])

以上，介绍了标量聚合函数和表值聚合函数的自定义实现，分别以具体的示例进行展示。特别需要提醒的是表值聚合函数自定义实现时针对emitValue和emitUpdateWithRetract方法的不同版本实现要求，该处在其官网上没有特别的说明，会导致运行异常，具体原因及解决办法在示例2emitUpdateWithRetract中有说明。

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

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

相关文章

CoRL 2023 获奖论文公布，manipulation、强化学习等主题成热门

CoRL 2023 获奖论文公布，manipulation、强化学习等主题成热门

今年大模型及具身智能领域有了非常多的突破性进展&#xff0c;作为机器人学与机器学习交叉领域的全球顶级学术会议之一&#xff0c;CoRL也得到了更多的关注。 CoRL 是面向机器人学习的顶会&#xff0c;涵盖机器人学、机器学习和控制等多个主题&#xff0c;包括理论与应用。今年…
阅读更多...
初探地理编码（2023.11.12）

初探地理编码（2023.11.12）

地理编码相识 2023.11.12 引言1、地理编码简介2、地理编码API和服务&#xff08;解决方案供应商 / 厂商&#xff09;2.1 高德2.2 百度2.3 超图2.4 天地图2.5 ArcGIS2.6 MapBox2.7 Cesium2.8 MapLocation 3、python实例3.1 pip安装依赖库&#xff08;python 3.6&#xff09;3.2 …
阅读更多...
MVVM框架：图片加载有问题

MVVM框架：图片加载有问题

一、前言&#xff1a;在我使用ImageView加载图片的时候添加如下代码发现报错 app:imageUrl"{viewModel.observableField.assetImg}"报错如下错误 二、原因&#xff1a;是啥我不太清楚好像是没有imageView的适配器&#xff0c;后来我看了一下确实没有 public class I…
阅读更多...
AD9371 Crossbar 和 I、Q数据 映射JESD204B传输层

AD9371 Crossbar 和 I、Q数据 映射JESD204B传输层

AD9371 系列快速入口 AD9371ZCU102 移植到 ZCU106 &#xff1a; AD9371 官方例程构建及单音信号收发 ad9371_tx_jesd -->util_ad9371_xcvr接口映射&#xff1a; AD9371 官方例程之 tx_jesd 与 xcvr接口映射 AD9371 官方例程 时钟间的关系与生成 &#xff1a; AD9371 官方…
阅读更多...
linux 安装 mini conda，linux下安装 Miniconda

linux 安装 mini conda，linux下安装 Miniconda

下载地址 https://docs.conda.io/projects/miniconda/en/latest/index.html 安装conda mkdir -p ~/miniconda3 wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh -O ~/miniconda3/miniconda.sh bash ~/miniconda3/miniconda.sh -b -u -p ~/mini…
阅读更多...
微信开发者工具如何使用

微信开发者工具如何使用

首先是下载微信开发者工具 链接: https://pan.baidu.com/s/1ri-fRCUQsz9vLxbTqtdPUQ 提取码: 8rhn 复制这段内容后打开百度网盘手机App&#xff0c;操作更方便哦 安装完打开就是以下界面 接下来进入正题 第一步新建或导入 新建 获取AppID 第一步&#xff1a;通过微信公众平…
阅读更多...
第12章 PyTorch图像分割代码框架-3：推理与部署

第12章 PyTorch图像分割代码框架-3：推理与部署

推理模块 模型训练完成后&#xff0c;需要单独再写一个推理模块来供用户测试或者使用&#xff0c;该模块可以命名为test.py或者inference.py&#xff0c;导入训练好的模型文件和待测试的图像&#xff0c;输出该图像的分割结果。inference.py主体部分如代码11-7所示。 代码11-7 …
阅读更多...
vue做的一个一点就转的转盘（音乐磁盘），点击停止时会在几秒内缓慢停止，再次点击按钮可以再次旋转，

vue做的一个一点就转的转盘（音乐磁盘），点击停止时会在几秒内缓慢停止，再次点击按钮可以再次旋转，

先看效果&#xff1a; 代码&#xff1a;主要部分我会红线画出来 css:部分&#xff1a; 源码&#xff1a; vue部分&#xff1a; <template><div class"song-lyric"><div><div class"type"><div class"right">&l…
阅读更多...
如何用自然语言 5 分钟构建个人知识库应用？我的 GPTs builder 尝试

如何用自然语言 5 分钟构建个人知识库应用？我的 GPTs builder 尝试

开发者的想象力闸门一旦打开&#xff0c;迎接我们的必然是目不暇接的 AI 应用浪潮冲击。 兴奋 早晨&#xff0c;我突然发现 ChatGPT 最新的 Create GPTs 功能可以用了。 这太让我意外了&#xff0c;没想到这么快。根据页面上的提示&#xff0c;我一直以为还得等上一周左右。于是…
阅读更多...
3C制造RFID产线智能化升级改造设计方案

3C制造RFID产线智能化升级改造设计方案

3C行业需求 近年来&#xff0c;随着政策的支持、相关技术的进步以及市场需求的推动&#xff0c;3C行业迅速发展&#xff0c;我国的3C市场已经进入了稳定发展阶段&#xff0c;作为仅次于汽车产业的大市场&#xff0c;3C产业在智能制造的推动下&#xff0c;越来越多的物联网技术…
阅读更多...
软考网络工程师知识点总结（四）

软考网络工程师知识点总结（四）

目录 61、FTP文件传输服务 62、DHCP动态主机配置协议 63、电子邮件服务 64、各种新技术相关概念及功能的考查 65、Windows的ipconfig命令 66、Windows的其它命令及说明 67、Linux系统关机和重启命令 68、Linux系统文件属性命令 69、SNMP协议版本 70、SNMP协议的报文类…
阅读更多...
【OpenCV(3)】linux arm aarch 是 opencv 交叉编译与使用

【OpenCV(3)】linux arm aarch 是 opencv 交叉编译与使用

文章目录 1、直接找github 别人编译好的2、自主编译参考 3使用CMake检查 参考 1、直接找github 别人编译好的 测试很多&#xff0c;找到一个可用的。 https://github.com/dog-qiuqiu/libopencv 它用了超级模块&#xff01; OpenCV的world模块也称为超级模块&#xff08;supe…
阅读更多...
Zeitgeist ZTG Token以及其预测市场加入Moonbeam生态

Zeitgeist ZTG Token以及其预测市场加入Moonbeam生态

波卡上的首选多链开发平台Moonbeam宣布与Zeitgeist达成XCM集成&#xff0c;将ZTG Token引入Moonbeam。此集成将使波卡内的Moonbeam和Zeitgeist网络之间的流动性得以流动&#xff0c;并通过Moonbeam的互连合约实现远程链集成。 Zeitgeist是一个基于波卡的Substrate区块链框架构…
阅读更多...
5个最佳开源RPA框架

5个最佳开源RPA框架

在最近两年中&#xff0c;RPA加上AI&#xff0c;即智能自动化流程&#xff0c;已经成为频繁讨论的话题&#xff0c;特别是在企业和机构的数字化转型过程中。自动化与智能化成为了提高效率的关键手段&#xff0c;而RPA便是迈向这一未来的起始步骤。 可以将RPA视为人体的躯干神经…
阅读更多...
[MySQL] MySQL库的基础操作

[MySQL] MySQL库的基础操作

文章目录 一、数据库的创建 1、1 库的创建 1、2 字符集与校验规则 1、2、1 查看字符集与校验规则 1、2、2 字符集与校验规则的设置 1、2、3 校验规则对数据库的影响 二、数据库的操作 2、1 查看数据库 2、2 删除数据库 2、3 修该数据库 2、4 数据库删除和备份 2、5 显示创建语…
阅读更多...
C#医学检验室(LIS)信息管理系统源码

C#医学检验室(LIS)信息管理系统源码

LIS:实验室信息管理系统 (Laboratory Information Management System简称:LIS)。 LIS 是面向医院检验科、检验中心、动物实验所、生物医疗研究所等科研单位研发的集数据采集、传输、存储、分析、处理、发布等功能于一体的信息管理系统。 一、完善的质控&#xff1a; 从样本管理…
阅读更多...
芯片设计工程师必备基本功——《设计与验证：Verilog HDL》

芯片设计工程师必备基本功——《设计与验证：Verilog HDL》

Verilog HDL 作为两大硬件描述语言之一&#xff0c;拥有很大的用户群。据调查&#xff0c;目前美国有 90%左右的 IC 设计人员使用 Verilog. 在中国&#xff0c;大概再 50% 左右的人在使用 Verilog 。 大量高校毕业生和部分软件设计人员正在不断涌入这个领域。要想尽快在 IC设计…
阅读更多...
Python开源项目CodeFormer——人脸重建（Face Restoration），模糊清晰、划痕修复及黑白上色的实践

Python开源项目CodeFormer——人脸重建（Face Restoration），模糊清晰、划痕修复及黑白上色的实践

无论是自己、家人或是朋友、客户的照片&#xff0c;免不了有些是黑白的、被污损的、模糊的&#xff0c;总想着修复一下。作为一个程序员 或者 程序员的家属&#xff0c;当然都有责任满足他们的需求、实现他们的想法。除了这个&#xff0c;学习了本文的成果&#xff0c;或许你还…
阅读更多...
git02-＞gui图形化界面使用,ssh协议,idea集成GIT

git02-＞gui图形化界面使用,ssh协议,idea集成GIT

gui图形化界面使用ssh协议idea集成GIT 1.gui图形化界面使用 2.ssh协议 git/github生成密钥并通过 操作分为本地电脑配置和github网站配置 第一步&#xff1a;本地电脑配置 右键空白处&#xff0c;选择Git Bash Here打开相关命令窗口 1.配置用户名和邮箱&#xff08;如果已经配…
阅读更多...
node插件MongoDB（三）—— 库mongoose 的使用和数据类型（一）

node插件MongoDB（三）—— 库mongoose 的使用和数据类型（一）

前言 提示&#xff1a;使用mongoose 的前提是你安装了node和 MongoDB。 mongoose 官网文档&#xff1a;http://mongoosejs.net/docs/index.html 文章目录 前言一、安装二、基本使用1. 打开bin目录的mongod.exe文件2. 基本使用的代码&#xff08;连接mongodb 服务&#xff09;3.…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023