营销网站建设网络公司,如何管理wordpress网站模板,网站备案 电信,常用wap网站开发工具 手机网站制作软件我们非常高兴地宣布#xff1a;RisingWave 1.10 版本正式发布#xff01;新版本为大家带来了许多重要更新#xff0c;例如#xff1a;新增用户自定义聚合函数 (UDAF)、支持从游标获取多个更新、支持可溢出哈希 Join、增强 CDC 连接器、新增 Sink 连接器等。一起来了解本次更…我们非常高兴地宣布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 AGGREGATEEmbedded Python UDFsUse 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:0000: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 CDCIngest data from PostgreSQL CDCIngest 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 设置为 5RisingWave 将间隔 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 decouplingSink data from RisingWave to Delta LakeSink 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。
更多细节请查看
Sink data from RisingWave to Amazon DynamoDBSink data from RisingWave to SQL ServerSink data from RisingWave to OpenSearch 8. 总结
以上只是 RisingWave 1.10 版本新增的部分功能如果您想了解本次更新的完整列表请查看更详细的发布说明。
9. 关于 RisingWave
RisingWave 是一款开源的分布式流处理数据库旨在帮助用户降低实时应用的开发成本。RisingWave 采用存算分离架构提供 Postgres-style 使用体验具备比 Flink 高出 10 倍的性能以及更低的成本。
加入 RW 社区欢迎关注公众号RisingWave 中文开源社区
想要了解和探索 RisingWave欢迎浏览我们的官网risingwave.com/
快速上手 RisingWave欢迎体验入门教程github.com/risingwave
深入理解使用 RisingWave欢迎阅读用户文档zh-cn.risingwave.com/docs
