CSA1.4：支持SQL流批一体化

大数据杂货铺

发布于 2021-07-02 10:22:02

6660

发布于 2021-07-02 10:22:02

文章被收录于专栏：大数据杂货铺

2020 年 10 月 Cloudera 收购了 Eventador，Cloudera Streaming Analytics (CSA) 1.3.0 于 2021 年初发布，该版本是从收购中合并 SQL Stream Builder (SSB) 的第一个版本，它将丰富的 SQL 处理带到已经很强大的 Apache Flink 产品中。

团队的重点转向将 Flink 数据定义语言( DDL) 和批处理接口带入 SSB。我们希望利用最新的上游 Flink 版本中令人兴奋的发展，并为 SSB 添加重要的功能。对于客户而言，这在 Cloudera 堆栈中开辟了大量新机会，以将现有数据足迹与流媒体数据源相结合。

我们很高兴地宣布 CSA 1.4.0 – 具有统一的流媒体和批处理界面。我们相信这项新功能将为物联网、金融、制造等领域的用例开启全新的功能。这使客户能够创建独特的 ETL 流、实时数据仓库和创建有价值的数据源，而无需大规模重新设计基础设施。

为什么是批处理+流媒体？

长期以来，我们一直被告知批处理和流（有界和无界系统）是正交技术——一种参考架构，其中流媒体为数据湖提供养料，仅此而已。其中批处理用于检查流的有效性（lambda），或者我们需要将所有内容都考虑为流（kappa）。

但在战壕中，作为数据从业者，我们想要更多。我们希望能够在我们的逻辑中处理批处理源和流媒体源，并拥有工具和 (SQL) 语法来轻松处理它们。我们希望能够以简单的方式轻松整合现有企业数据源和高速/低延迟数据流。我们需要灵活地处理批处理 API 和流 API 以及无缝读取和写入它们的连接性。我们需要进行试验、迭代，然后部署无需大量数据重放即可扩展和恢复的处理器。我们希望在可能的情况下自动推断模式，并在需要时使用丰富的工具来构建它们。

最终，业务并不关心源数据的形式，我们需要一个框架来快速轻松地交付数据产品，而无需添加大量基础设施或需要下游数据库。这种架构没有一个花哨的名字——主要是因为它应该一直是这样运作的。因此，CSA 1.4 使构建这些数据产品变得轻而易举。

Flink 的一点历史

Cloudera Steaming Analytics 由 Apache Flink 提供支持，包括 SQL Stream Builder 和核心 Flink 引擎。但是，您可能不知道 Apache Flink 从一开始就是一个批处理框架。然而，Flink 很早就通过两个独立的 API 接受了批处理和流媒体。Flink 改进提案 131重新定义了 Flink API，重点关注同一 API 下有界/无界处理的统一。以前，必须选择一个API或另一个 API。随着 Flip-131 的引入，处理模式将完全从 table API 下的程序中抽象出来——允许开发人员编写将两种处理范式巧妙结合的程序。Flink 一直专注于正确的结果，并支持一次处理。将引擎的强大功能与限时连接语法相结合，为我们提供了使用简单连接语法查询有界和无界数据的选项。这是一个完整而彻底的游戏规则改变者。

SQL Stream Builder 满足有界查询

与 Flink 本身相比，SQL Stream Builder 最初是作为一个纯粹的流接口。从 CSA 1.4 开始，SSB 允许运行查询以连接和丰富来自有界和无界源的流。SSB 可以从 Kudu、Hive 和 JDBC 源加入以丰富流。随着时间的推移，我们将继续添加更多有界的源和接收器。SSB 一直能够加入多个数据流，但现在它也可以通过批处理源进行丰富。

数据定义语言 (DDL)

新功能的核心是将 Flink DDL 并入 SSB。表是用Schema（推断的或指定的）定义的，然后可以像任何其他源一样在它们上运行连续 SQL。此外，可以自动访问 Cloudera Data Platform 中的源。

-- auto-inferred from CDP catalog import
CREATE TABLE `CDP_Hive_Catalog`.`airplanes`.`faa_aircraft` (
  `tailnumber`  VARCHAR(255),
  `model`  VARCHAR(255),
  `serial`  VARCHAR(255),
  `icao`  VARCHAR(255),
  `owner`  VARCHAR(255)
) WITH (
  ...
)

读取和丰富批处理数据

例如，我们在这里丰富了测量制造系统测试状态的数据流。我们用来自 (b) 的员工数据丰富了流 (a)。我们利用 Flink 语法为表指定时间（proctime()），并指定一个连接键。

SELECT
  a.stationid, a.test, b.managername
FROM
  mfgrline AS a
  JOIN `CDP_Kudu_Catalog`.`HR_db`.`impala::mfgr.stations`
  FOR SYSTEM_TIME AS OF PROCTIME() AS b
  ON a.stationid = b.stationid

还可以在一个语句中加入多个源，包括流到流的连接：

SELECT
 geo_event.eventTimestamp, geo_event.driverId, geo_event.eventTime,geo_event.eventSource,
 geo_event.truckId,geo_event.driverName,geo_event.routeId,geo_event.route,geo_event.eventType,
 geo_event.latitude, geo_event.longitude, geo_event.correlationId, geo_event.geoAddress,
 speed_event.speed,
 driver.certified, driver.wage_plan,
 timesheet.hours_logged, timesheet.miles_logged       
FROM
  geo_events_json AS geo_event
  JOIN speed_events_json AS speed_event
    ON (geo_event.driverId = speed_event.driverId)
  LEFT JOIN CDP_Hive_Catalog.employees_hr_hive_db.driver
    FOR  SYSTEM_TIME AS OF PROCTIME() driver
    ON driver.driverid = geo_event.driverId
  LEFT JOIN `CDP_Kudu_Catalog`.`default_database`.`impala::employees_hr_kudu_impala_db.timesheet`
      FOR  SYSTEM_TIME AS OF PROCTIME() timesheet
      ON (timesheet.driverid = geo_event.driverId  AND timesheet_week = 1)     
WHERE
  geo_event.eventTimestamp BETWEEN
    speed_event.eventTimestamp - INTERVAL '1' SECOND AND
    speed_event.eventTimestamp + INTERVAL '1' SECOND
  AND geo_event.eventType <> 'Normal'
  AND driver.wage_plan = 'hours'
  AND timesheet.hours_logged > 45

写入批处理系统

SSB 也可以作为接收器写入批处理系统。这不仅可以用于存储某些计算的结果，还可以保持计算的逻辑状态。例如，为您因欺诈而关闭的帐户保留分类帐 - 这样您就不会重新发送未来的请求。要写入接收器，就像定义一个表并将其选择为接收器一样简单。

解锁新的用例和架构

借助 CSA 1.4 提供的新功能，新的用例以及降低延迟和加快上市时间的新功能成为可能。

分布式实时数据仓库——通过物化视图将流数据作为事实与批量数据作为维度进行连接。例如，执行丰富的点击流分析，或将传感器数据与历史测量值结合起来。SSB 中的新 DDL 接口提供了从CDP 堆栈中的任何位置定义流和批处理源并使用连续 SQL 连接它们的功能。
数据科学——分析需要上下文。例如，通过使用笔记本中 Python 模型的历史记录丰富行为流，为客户实时提供个性化体验。SQL Stream Builder 为物化视图提供了一个简单的 REST 接口，可以轻松地与笔记本内的 Python 和 Pandas 集成- 因此数据科学家可以专注于本地工具中的小而有价值的数据集，而不必解析流数据的流水。
实时制造能力——在制造中，能够无缝地处理来自整个企业的数据源，然后实现仪表板的视图可以消除浪费、控制成本并提高质量。例如，将历史站故障率与当前遥测相结合，以在Cloudera Dataviz或 Grafana 中显示预测输出。