当前位置：首页 > news >正文

RisingWave 1.10 发布！新增用户自定义聚合函数

news 2025/7/8 1:22:13

我们非常高兴地宣布：RisingWave 1.10 版本正式发布！新版本为大家带来了许多重要更新，例如：新增用户自定义聚合函数 (UDAF)、支持从游标获取多个更新、支持可溢出哈希 Join、增强 CDC 连接器、新增 Sink 连接器等。一起来了解本次更新的主要亮点吧！

1. 新增 UDAF

现在，您可以用 Python 和 JavaScript 创建嵌入式用户自定义聚合函数（User-defined Aggregate Function）。这些函数在 RisingWave 中定义，并使用嵌入式解释器执行。定义后，您可以像使用任何内置 SQL 聚合函数一样使用它们。创建 UDAF 需要使用 CREATE AGGREGATE 命令，其一般语法如下：

CREATE AGGREGATE function_name ( argument_type [, ...] )RETURNS return_typeLANGUAGE language_nameAS $$ function_body $$;

其中，函数体 (function_body) 内是一系列返回聚合值的函数，可以用 Python 或 JavaScript 定义。对于 UDAF，您必须定义 create_state 函数，它会创建一个新状态 (State)，用于维护聚合函数的持续计算，促成高效的计算结果。您还必须定义 accumulate 函数，它更新并返回当前状态值。此函数将状态和来自聚合函数定义的输入参数作为参数。

此外，您可以选择定义 finish 函数，该函数返回聚合函数的结果，定义时必须将状态作为输入参数。如果您未定义此函数，则函数将返回当前状态。您还可以选择 retract 函数，它会撤回当前状态的值，然后返回该值。

UDAF 赋予了 RisingWave 更复杂的计算能力，让您在处理数据时更灵活更自主。

更多细节，请查看：

CREATE AGGREGATE
Embedded Python UDFs
Use UDFs in JavaScript

2. 从游标获取多个更新

在 v1.9 中，我们为订阅引入了子脚本和游标，允许您检索对表或物化视图所做的更新。以前，您只能使用 FETCH 命令逐行从游标中检索更新。现在，您可以指定从游标中检索多少行。以下 SQL 查询从游标 cur1 中检索四个最新的更新。

FETCH 4 FROM cur1;

此功能更方便您查看表和物化视图的最近更改。此外，结果表的列名已更新为与源表或物化视图的列名匹配，此前的格式为 table_name.col_name。

更多细节，请查看：

Fetch from cursor

3. 支持可溢出哈希 Join

为提高 RisingWave 在 Join 两个大表时的性能，我们现在支持可溢出哈希 Join。目前，RisingWave 正在使用的是哈希 Join，哈希表在内存中构建，它们能够很好地并行化和扩展，但需要大量内存去构建，当表很大时，可能会导致内存不足问题。可溢出哈希 Join 解决了这个问题，在 Join 查询期间内存使用量高时，RisingWave 可以利用磁盘空间。

4. 对 CDC Source 连接器的增强

本次版本中，我们继续改进了现有 CDC Source 连接器，为您提供更流畅的流处理体验。此版本包含两个新功能：自动映射 Schema 和元数据列。但请注意，这些新功能并不适用于所有 CDC 连接器，因此请继续阅读以了解更多详情。

4.1 自动映射 Schema

在创建 MySQL 或 PostgreSQL CDC 表时，RisingWave 现在会自动将上游表的 Schema 映射到 RisingWave 表。创建表时可以使用 * 以从源表中导入所有列，而无需单独定义列。但是，如果在表创建过程中指定了其他列，则不能使用 *。

让我们用一个简单例子说明这个过程。首先，我们用以下 SQL 查询连接到 MySQL 数据库。在从 MySQL 或 PostgreSQL 导入 CDC 数据时，您必须先创建一个 Source，用于连接到数据库，然后再从各个表中导入数据。

CREATE SOURCE mysql_source WITH (connector = 'mysql-cdc',hostname = '127.0.0.1',port = '3306',username = 'root',password = 'password',database.name = 'mydb',server.id = 5888
);

接下来，我们创建一个表，从 MySQL 数据库中上游表 tbl1 中导入所有列。mysql_tbl 的列将对应 tbl1 的列。

CREATE TABLE mysql_tbl (*)
FROM mysql_source TABLE 'mydb.tbl1';

此功能使在 RisingWave 中创建 CDC 表更加高效。

4.2 包含元数据列

在创建 MongoDB、MySQL 或 PostgreSQL CDC 表时，可以使用 INCLUDE 子句附加元数据列。如果需要将元数据列添加到已有的 CDC 表中，则需要在 RisingWave 中重新创建该表。

对于 MongoDB、MySQL 和 PostgreSQL CDC 表，您可以使用 INCLUDE 子句导入上游提交时间戳。对于历史数据，默认填充数据为 1970-01-01 00:00:00+00:00。

对于 MongoDB，您可以使用 INCLUDE 子句导入 collection_name。
对于 MySQL 和 PostgreSQL，您可以导入 database_name、schema_name 和 table_name。

INCLUDE 子句的语法如下：

INCLUDE metadata_col [AS col_name];

metadata_col 可以是上述提到的任何元数据列。在表 Schema 定义之后，此子句可以在创建表时使用。

以下是一个示例，从 MySQL 表中导入元数据列 timestamp 和 database_name。

CREATE TABLE tbl_meta (id int,name varchar,age intPRIMARY KEY (id)
) INCLUDE TIMESTAMP AS commit_ts
INCLUDE DATABASE_NAME AS db_name
FROM mysql_source TABLE 'mydb.tbl2';

更多细节，请查看：

Ingest data from MySQL CDC
Ingest data from PostgreSQL CDC
Ingest data from MongoDB CDC

5. 增强现有 Sink 连接器

5.1 默认 Sink 解耦

对于 ClickHouse、Google Pub/Sub、Kafka、Kinesis、MQTT、NATS 和 Pulsar Sink 连接器，Sink 解耦将默认启用。之前，此功能只在 Sink 是 append-only 时才会启用，现在则不再有此限制。Sink 解耦会在 RisingWave 和下游系统之间插入一个缓冲队列，以确保 RisingWave 不受下游系统性能问题的影响。

如果您想禁用 Sink 解耦，请使用会话变量 sink_decouple。

SET sink_decouple = false;

5.2 检查点解耦选项

对于 Delta Lake 和 StarRocks Sink 连接器，您可以使用 commit_checkpoint_interval 参数，将下游系统的 commit 与 RisingWave 的 commit 解耦。这意味着，RisingWave 将在达到指定的检查点间隔时提交数据，而不是在每个屏障处提交数据。

例如，如果 commit_checkpoint_interval 设置为 5，RisingWave 将间隔 5 个检查点提交一次数据。这可以减少生成的目标表版本，提升查询性能。

在创建 Delta Lake 或 StarRocks Sink 连接器时，commit_checkpoint_interval 参数应在 WITH 选项中指定。

CREATE SINK s1_sink FROM s1_source
WITH (connector = 'deltalake',type = 'append-only',location = 's3a://my-delta-lake-bucket/path/to/table',s3.endpoint = '<https://s3.ap-southeast-1.amazonaws.com>',s3.access.key = '${ACCESS_KEY}',s3.secret.key = '${SECRET_KEY}',commit_checkpoint_interval = 5
)

更多细节，请查看：

Sink decoupling
Sink data from RisingWave to Delta Lake
Sink data from RisingWave to StarRocks

6. 新增 Sink 连接器

RisingWave 一直在持续添加下游系统连接器，扩展其生态系统。我们现在支持将数据 Sink 到 DynamoDB 和 Microsoft SQL Server。如果您对特定连接器感兴趣，请参阅我们的集成页面。您可以投票以表示对特定连接器感兴趣，或在其可用时收到通知。

6.1 Amazon DynamoDB

Amazon DynamoDB 是一个 NoSQL 数据库，旨在处理高容量的结构化和半结构化数据。它提供一致的高性能和易扩展性。要将数据从 RisingWave Sink 到 DynamoDB 表，需使用 CREATE SINK 命令。Sink 到 DynamoDB 表时，您的 RisingWave 源表必须有一个由两列组成的复合主键。它们需要对应 DynamoDB 目标表中定义的分区键和排序键。

例如，如果您想 Sink 到名为 books_dynamo 的 DynamoDB 表，该表具有分区键 isbn 和排序键 edition，则 RisingWave 表 Schema 应定义如下：

CREATE TABLE IF NOT EXISTS books_rw (isbn varchar,edition int,title varchar,author varchar,primary key (isbn, edition)
);

然后再创建 Sink 连接器，将数据从 books_rw Sink 到 books_dynamo。

CREATE SINK dynamo_sink
FROM movies
WITH (connector = 'dynamodb',table = 'books_dynamo',primary_key = 'isbn, edition',endpoint = '<http://localhost:8000>',region = 'region,access_key = 'access_key',secret_key = 'secret_key'
);

6.2 Microsoft SQL Server

Microsoft SQL Server 是一个强大的关系数据库管理系统，支持广泛的数据事务处理、商业智能等功能。它使用 T-SQL，并包括 SQL Server 集成服务、报告服务和分析服务等工具。RisingWave 支持将数据 Sink 到自托管的 SQL Server 和 Azure SQL。

以下是一个示例，我们创建了 Sink 连接器 sqlserver_sink，将数据从物化视图 mv1 Sink 到 SQL Server 表 sqlserver_tbl。因为这是一个 Upsert Sink ，我们定义了主键 pk1 和 pk2。

 CREATE SINK sqlserver_sinkFROM mv1WITH (connector = 'sqlserver',type = 'upsert',sqlserver.host = 'sqlserver-server',sqlserver.port = 1433,sqlserver.user = 'user',sqlserver.password = 'password',sqlserver.database = 'mydb',sqlserver.table = 'sqlserver_tbl',primary_key = 'pk1, pk2',
);

6.3 OpenSearch

OpenSearch 是一个开源的搜索和分析引擎，旨在实时搜索、分析和可视化大量数据。它源自 ElasticSearch，适用于日志和事件数据分析、企业搜索、监控观测等各种应用。

要将数据从 RisingWave Sink 到 OpenSearch，您可以使用 CREATE SINK 命令。

CREATE SINK opensearch_sink
FROM table1
WITH (connector = 'opensearch',index = 'id1',primary_key = 'types_id',url = '<http://opensearch:8080>',username = 'user',password = 'password'
);

更多细节，请查看：

Sink data from RisingWave to OpenSearch

7. 保留内存算法变更

现在，用于计算默认保留内存 (Reserved Memory) 的算法已更改。保留内存用于为 RisingWave 提供调整内存使用量的缓冲时间，以应对输入数据的额外涌入。之前，我们将计算节点总内存的 20% 用作保留内存。现在，保留内存的计算方式为：前 16GB 内存的 30% + 剩余内存的 20%。通过这种计算方法，保留内存可以根据您的设置进行扩展，更好地平衡系统性能和内存利用率。

如果此方法不适合您，您可以使用启动选项 --reserve-memory-bytes 或环境变量 RW_RESERVED_MEMORY_BYTES 指定保留内存，但需要注意，保留内存必须至少为 512MB。

更多细节，请查看：