2021 年 3 月，StreamNative 宣布与Apache Flink 创始开发人员组建的 Ververica 公司成为云合作伙伴。该合作促成了 StreamNative 与 Ververica 的密切协作并加强了对 Apache Pulsar 和 Apache Flink 的整合。StreamNative 借此合作加速打造基于 Apache Pulsar 和 Apache Flink 的实时数据洞察平台。

StreamNative 近期正式在 StreamNative Cloud 上推出 Flink SQL 功能。StreamNative Cloud 上的 Flink SQL 提供了一个直观的交互式 SQL 界面，可降低在 Apache Pulsar 上构建实时数据查询的复杂性。

选择 Apache Flink 与 Flink SQL

Apache Flink 是一种分布式流数据处理引擎，可提供高吞吐量、低延迟的数据处理，以及强大的抽象和操作灵活性。用户通过 Apache Flink 可以轻松开发和部署事件驱动的应用程序、数据分析作业和数据管道，以处理复杂分布式系统中的实时和历史数据。Apache Flink 拥有强大的功能和成熟的社区，已被诸多全球最知名的数据驱动企业广泛采用，如阿里巴巴、Netflix 和 Uber。

Flink SQL 为存储在 Apache Pulsar 中的事件提供了关系抽象，支持流批一体处理场景下的 SQL 标准查询。使用 Flink SQL，用户无需编写 Java 或 Python 代码，即可以编写 SQL 查询并洞察实时数据。

Apache Flink 和 Flink SQL 凭借强大的执行引擎和简单的抽象层，提供了一种低成本的开发、维护分布式实时数据处理的解决方案。通过集成 Pulsar 和 Flink，StreamNative 可为客户提供同时进行流存储和流计算的完整流式解决方案。

StreamNative Cloud——基于 Flink + Pulsar 的云原生流数据平台

关于 StreamNative Cloud

StreamNative Cloud 是 StreamNative 推出的基于 Apache Pulsar 的云端服务产品。该产品的发布，让 Pulsar®-as-a-Service（PaaS）成为现实 ，加速 Pulsar 推广与落地，助力企业推进“流优先”策略。

StreamNative Cloud 提供的 “PaaS” 服务，操作简单、快速、可靠，效率高、成本低。对于想要搭建下一代消息和事件流系统服务的企业来说，StreamNative Cloud 是首选的解决方案。

使用 StreamNative Cloud，开发者无需在本地维护复杂的服务系统，可以更专注于核心业务应用、微服务开发等。StreamNative Cloud 支持用户直接在云端开启 Apache Pulsar 之旅，助力企业快速落地“流优先”策略，加速业务发展。

批流一体

企业对实时数据洞察的需求至关重要，但数据洞察力不仅限于实时数据。企业也需要整合并理解大量历史数据，以便全面了解其业务。因此要求系统也能够捕获、存储和计算实时和历史数据。

Pulsar 的分层存储模型^[1]提供了批处理和流处理所需的存储能力，使 StreamNative Cloud 能够提供批流统一的存储。Apache Flink 和 Flink SQL 的集成提供了统一的批处理和流处理，而 Flink SQL 简化了执行。

在流优先的世界中，数据的核心抽象是无限流。下图中的表是从流中派生出来的，并随着新数据到达流中而不断更新。Apache Pulsar 提供无限流的存储，Apache Flink 创建流式表（streaming table）形式的物化视图的引擎。用户可以运行流查询来执行连续转换，或对流式表（streaming table）运行批处理查询以实时获取流中每个键的最新值。

流与表

集成 Apache Flink 与 Apache Pulsar，可以帮助企业能够以新的方式表达和处理流数据。Pulsar 无限流是流式数据的核心存储抽象，其他一切都是无限流的物化视图，包括数据库、搜索索引或企业中的其他数据服务系统。在 StreamNative Cloud 上，用户可以使用 Apache Flink 以流式方式创建这些派生视图所需的所有数据丰富（Data Enrichment）和 ETL。通过将历史上下文与实时数据分析相结合，可以实时完成监控、安全、异常和威胁检测、分析和故障响应。

StreamNative Cloud 提供完整的流数据解决方案

Flink SQL 使用场景

在使用 StreamNative Cloud 上的 Flink SQL 时，Pulsar 集群被视为 Flink Catalog。用户可以使用 Flink SQL 查询 Apache Pulsar 中的无限事件流。以下是在 Pulsar 流上利用流式 SQL 查询的一些主要场景：

实时数据监控

我们经常认为监控是使用计数器和仪表来跟踪低级性能统计信息。虽然这些指标可显示 CPU 使用率很高，却无法显示应用程序是否在做它应该做的事情。Flink SQL 允许用户从应用程序生成的消息流中定义自定义指标，如记录事件、捕获的更改数据等等。例如，在云服务中需要检查每次新用户注册时是否发送欢迎电子邮件、创建新用户记录以及对他们的信用卡收费，这些功能可能分布在多个不同的服务或应用程序上，并且用户通常希望监视特定 SLA 内每个新客户的行为。

下面是一个流式 SQL 查询示例，用于监视错误代码流上的错误计数。

INSERT INTO error_countsSELECT error_code, count(*) FROM monitoring_streamGROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), error_codeHAVING type = ‘ERROR’;

实时异常检测

在安全场景中，开发人员不是监视应用程序行为或业务行为，而是寻找欺诈、滥用、垃圾邮件、入侵或其他不良行为的模式。Flink SQL 提供了一种简单且实时的方式来定义这些模式并查询实时 Pulsar 流。

下面是一个流式 SQL 查询示例，用于检测交易流中的欺诈行为。

INSERT INTO possible_fraudSELECT card_number, count(*)FROM transactionsGROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), card_numberHAVING count(*) > 3;

实时数据管道

企业构建实时数据管道以丰富数据。这些数据管道捕获、转换并连接来自多个数据库的数据变动，并将它们存储在键值数据库、搜索索引、缓存或其他数据服务系统中。

长期以来，企业多用 ETL 管道进行定期的批处理作业，例如实时摄取原始数据，然后每隔几个小时对其进行转换以实现高效查询。然而在很多实时场景中，例如交易或支付处理场景，这种延迟是不可接受的。Flink SQL 与 Pulsar I/O 连接器一起实现了不同系统之间的实时数据集成。

在 StreamNative Cloud 中，可以使用连接将元数据存储在不同的表中来扩充事件流，或者在将流加载到另一个系统之前对个人身份信息 (PII) 数据执行简单的过滤。

下面的流式 SQL 查询示例，用于使用用户表扩充点击流。

INSERT INTO vip_usersSELECT user_id, page, actionFROM clickstream cLEFT JOIN users u ON c.user_id = u.user_idWHERE u.level = ‘Platinum’;

Flink SQL 中的 Pulsar 抽象

Flink SQL 和 Apache Pulsar 的集成利用 Flink Catalog API^[2] 来引用现有的 Pulsar 元数据，并自动将它们映射到 Flink 对应的元数据。此集成中有一些核心抽象映射到 Pulsar 中的核心抽象，并允许用户使用 SQL 操作 Pulsar topic。以下是一些核心的抽象概念：

•Catalog：Catalog 是数据库的集合。它映射到现有的 Pulsar 集群。•数据库：数据库是表的集合。它映射到 Apache Pulsar 中的命名空间。Pulsar 集群中的所有命名空间都会自动转换为 Pulsar Catalog中的 Flink 数据库。还可以通过数据定义语言 (DDL) 查询创建或删除数据库，同时创建或删除底层 Pulsar 命名空间。

CREATE DATABASE userdb;

•表：Pulsar topic 可以表示为 STREAMING 表或 UPSERT 表。•Schema: 如果一个 topic 上已经存在一个 schema，Pulsar topic schema 将自动映射为 Flink 表 schema。如果 Pulsar topic 不存在，通过 DDL 查询创建的表会将 Flink 表 schema 转换为 Pulsar schema 来创建 Pulsar topic。•元数据列：Pulsar 消息的元数据和属性将映射到 Flink 表的元数据列。这些元数据列如下：•messageId: Pulsar 消息的消息 ID。（只读）•sequenceId: Pulsar 消息的序列 ID。（只读）•publishTime: Pulsar 消息的发布时间戳。（只读）•eventTime: Pulsar 消息的事件时间戳。（可读/可写）•properties: Pulsar 消息的消息属性。（可读/可写）

在 Flink 中，Pulsar topic 可以表示为流式表（STREAMING table）或 UPSERT 表。

流式表——无界的结构化数据序列

流式表表示无界的结构化数据序列（即“事实”）。假设有一个金融交易流如“Jack 向 Kate 转账 100 美元，然后 Alice 向 Kate 转账 200 美元”。表中的事实是不可变的，这意味着可以将新事件插入表中，但永远不能更新或删除现有事件。Pulsar 命名空间中的所有 topic 将自动映射到配置为使用 Pulsar 连接器的 Catalog 中的流式表。流式表也可以通过 DDL 查询创建或删除，其中底层 Pulsar topic 将被创建或删除。

CREATE TABLE pageviews (    user_id BIGINT,    page_id BIGINT,    viewtime TIMESTAMP,    user_region STRING,    WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND);

UPSERT 表——事实进展

一个 upsert 表代表了一系列不断发展的事实。假设有一个包含最新财务信息的表，诸如 “Kate 的当前账户余额为 300 美元”，它相当于传统的数据库表，但通过流语义（例如窗口）进行了丰富。UPSERT 表中的事实是可变的，这意味着可以向表中插入新事实，并且可以更新或删除现有事实。可以通过将 connector 指定为 upsert-pulsar 来创建 upsert 表。

CREATE TABLE pageviews_per_region (    user_region STRING,    pv BIGINT,    uv BIGINT,    PRIMARY KEY (user_region) NOT ENFORCED) with(    “connector” = “upsert-pulsar”};

通过整合流式表和 upsert 表的概念，Flink SQL 允许将表示当前世界状态的 upsert 表与表示当前正在发生的事件的流式表连接起来。根据 topic 处理的预期语义，Pulsar topic 可以表示为 Flink SQL 中的流式表或 upsert 表。

例如，如果您想将 topic 中的数据作为一系列独立值读取，可以将 Pulsar topic 视为流式表。此类流式表的一个场景是捕获页面浏览事件的 topic，其中每个页面浏览事件不相关且彼此独立。另一方面，如果您想将 topic 中的数据作为不断整合的可更新值来读取，则可以将该 topic 视为 upsert topic。在 Flink 中应该作为 UPSERT 表读取的 topic 的一个场景是捕获用户元数据，其中每个事件代表特定用户 ID 的最新元数据，包括其用户名、地址或用户偏好。

深度解析 StreamNative Cloud 上的 Flink SQL

StreamNative Cloud 架构

StreamNative Cloud 在控制平面和云池（Cloud Pool）运行。

控制平面包括 StreamNative 在其自己的云帐户中管理的后端服务。后端服务主要包括 Cloud API 服务和 Cloud 控制台。用户可以通过 Cloud 控制台与 StreamNative Cloud 进行交互，应用程序可以通过 Cloud API 服务与其进行交互。

StreamNative 可以在自己的云帐户或客户的云帐户中管理云池，SQL 查询和 Pulsar 集群都在云池中运行。

下图演示了如何在 StreamNative Cloud 系统中实现身份验证/授权。这里假设数据已经被 StreamNative Cloud 上的 Pulsar 集群获取，但是用户可以使用 Pulsar 的发布/订阅消息 API 从外部数据源获取数据，例如事件数据、流数据、IoT 数据等。

用户或应用程序可以与 StreamNative 控制平面交互来创建 Pulsar 集群。一旦 Pulsar 集群准备就绪，用户可以创建一个 Flink 会话集群并使用 StreamNative 云控制台中的 SQL 编辑器来发起交互式查询，或者创建长时间运行的部署来持续处理 Pulsar 集群中的数据流。

Flink SQL 与 Pulsar 集群交互示意图

对于每个 Flink 会话集群，都有一个 SQL 网关进程解析 SQL 查询并在本地执行查询，或将查询提交给 Flink 集群。SQL 网关中的每个 SQL 会话都会启动 Pulsar Catalog，每个 Catalog 代表一个现有的 Pulsar 集群。该 Catalog 包含安全访问 Pulsar 集群所需的所有必要信息。DDL 查询直接在 SQL 网关中执行，而所有 DML 查询都会提交到 Flink 会话集群执行。所有 SQL 查询都被模拟为出于安全目的提交它们的实际用户。

使用 StreamNative Cloud

点击下方图片可获得视频示范。

注意：目前仅可在 StreamNative Cloud 付费集群上使用 Flink SQL 特性。

StreamNative Cloud Managed Service 正在提供限时优惠！扫描下方二维码或点击阅读原文联系我们，体验云上批流解决方案！

关于 StreamNative

StreamNative 是一家开源基础软件公司，由 Apache 软件基金会顶级项目 Apache Pulsar 创始团队组建而成，围绕 Pulsar 打造下一代云原生批流融合数据平台。StreamNative 作为 Apache Pulsar 商业化公司，专注于开源生态和社区构建，致力于前沿技术领域的创新，创始团队成员曾就职于 Yahoo、Twitter、Splunk、EMC 等知名大公司。

引用链接

[1] 分层存储模型: http://pulsar.apache.org/docs/en/concepts-tiered-storage/
[2] Flink Catalog API: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/catalogs.html