做书的网站有哪些内容,c2c电子商务的网站,做电影字幕的网站,微信管理系统登录入口概览
在构建事件驱动应用时#xff0c;人们面临着两大挑战#xff1a;1#xff09;低延迟处理大量数据#xff1b;2#xff09;实现流数据的实时摄取和转换。
结合 RisingWave 的流处理功能和 ScyllaDB 的高性能 NoSQL 数据库#xff0c;可为构建事件驱动应用和数据管道…概览
在构建事件驱动应用时人们面临着两大挑战1低延迟处理大量数据2实现流数据的实时摄取和转换。
结合 RisingWave 的流处理功能和 ScyllaDB 的高性能 NoSQL 数据库可为构建事件驱动应用和数据管道提供有效的解决方案。
RisingWave 是什么
RisingWave 是一款专为流处理设计且与 PostgreSQL 兼容的数据库。它擅长摄取实时数据流、执行各种转换并实现对结果的即时查询。
ScyllaDB 是什么
ScyllaDB 是一款高性能分布式 NoSQL 数据库擅长处理大量数据并为应用程序提供低延迟访问。ScyllaDB 与 Apache Cassandra 数据模型和协议兼容这意味着在许多情况下它可以作为 Cassandra 的简单替代品。ScyllaDB 还提供与 Amazon DynamoDB 兼容的 API。
ScyllaDB 具有低延迟和高吞吐量的特性适合为需要快速数据访问的实时应用提供服务如在线游戏、实时分析或物联网IoT应用。
二者协同
RisingWave 擅长处理流数据包括对流数据的摄取、连接和转换。而 ScyllaDB 则能以极低的延迟为实时应用提供大量数据。
这两个系统为构建事件驱动应用或管道提供了坚实的基础。RisingWave 可在事件发生时即时处理事件数据其内置的 ScyllaDB 连接器可实时将处理后的数据导出到 ScyllaDB。这种集成能够确保数据随时可供实时应用或管道查询使用。
如何将 RisingWave 与 ScyllaDB 集成
我们将使用以下示例来演示如何使用 RisingWave 和 ScyllaDB 构建事件驱动应用。设想一下电子商务中的个性化推荐场景通过连接点击流和产品目录流我们可以实时分析用户的偏好并提供个性化推荐。 连接点击流和产品目录流
点击流示例如下
{user_id: john_doe,item_id: 12345,timestamp: 2023-03-08T15:30:00Z
}产品目录流示例如下
{item_id: 12345,category: electronics,price: 100,timestamp: 2023-03-08T10:00:00Z
}步骤 1从 RisingWave 摄取 Kafka 的实时数据
假设我们已将这两个流的数据分别打包到两个 Kafka Topic 中。
现在在 RisingWave 中创建两个 Source 来摄取这两个流
# 为点击流创建一个 Source
CREATE SOURCE clickstream (user_id VARCHAR,item_id VARCHAR,timestamp TIMESTAMPTZ
)
WITH (type kafka,kafka_topic clickstream,kafka_brokers localhost:9092
);# 为产品目录流创建一个 SourceCREATE SOURCE product_catalog (item_id VARCHAR,category VARCHAR,price NUMERIC,timestamp TIMESTAMPTZ
)
WITH (type kafka,kafka_topic product_catalog,kafka_brokers localhost:9092
);步骤 2在 ScyllaDB 中创建表
由于我们要实时连接数据流并将数据导出到 ScyllaDB因此需要在 ScyllaDB 中创建一个表来保存连接后的数据流。
CREATE TABLE joined_stream (user_id TEXT,item_id TEXT,timestamp TIMESTAMPTZ,category TEXT,price DECIMAL,PRIMARY KEY (user_id, item_id, timestamp)
);步骤 3执行流 Join 并导出到 ScyllaDB
在 RisingWave 中您可以通过创建 Sink 将数据导出到下游系统。在 CREATE SINK 语句中还可以包含数据转换逻辑。创建 Sink 时与创建 Source 或者实时物化视图类似本质上都是创建了持续数据处理的任务。在 RisingWave 中如果要创建的是简单直接的实时数据处理管道只需两条 SQL 语句即可做到CREATE SOURCE 和 CREATE SINK。
CREATE SINK joined_stream AS
SELECT c.user_id, c.item_id, c.timestamp, p.category, p.price
FROM clickstream c
JOIN product_catalog p ON c.item_id p.item_id;
WITH (connectorcassandra,typeappend-only,cassandra.url node1,node2,node3,cassandra.keyspace keyspace,cassandra.table joined_stream
);有关详细的句法和参数信息请参阅 从 RisingWave 导出数据到 Cassandra 或 ScyllaDB。
到这一步ScyllaDB 中的数据已可为应用或下游系统提供查询服务。
总结
只需三个步骤我们就建立了无缝的连续数据处理管道自动执行流 Join 并将连接后的数据导出到 ScyllaDB。ScyllaDB 的高性能使得实时应用能够以低延迟查询数据。这一集成的独特之处在于整个工作流程的设置非常简单。
以上是一个简单的用例示范除此之外使用 RisingWave您可以轻松地过滤、连接和转换流数据轻松表达复杂的转换逻辑。我们鼓励您进一步探索如有任何疑问或需要支持请联系我们。
事件驱动应用和数据管道的价值正在不断增长能够轻松配置技术栈是一大优势。RisingWave 和 ScyllaDB 的集成可简化技术栈使您能够专注于通过实时数据处理和分析实现价值。
资源
有关 RisingWave 和 ScyllaDB 的可运行集成请查看此演示。使用 RisingWave 和 ScyllaDB 进行经济高效的流处理ScyllaDB 文档RisingWave 用例RisingWave 文档有关 RisingWave 所支持集成的完整列表请查看此页面。
