KSQL：Apache Kafka的流式SQL

更新：KSQL 现在可作为Confluent Platform的一个组件提供。

我很高兴地宣布KSQL，为Apache kafka流SQL引擎®。KSQL降低了流处理世界的入口，提供了一个简单而完全交互的SQL接口，用于处理Kafka中的数据。您不再需要使用Java或Python等编程语言编写代码！KSQL是分布式的，可扩展的，可靠的和实时的。它支持各种强大的流处理操作，包括聚合，连接，窗口化，会话化等等。

一个简单的例子

查询流数据甚至意味着什么，这与SQL数据库相比如何？

嗯，它实际上与SQL数据库完全不同。大多数数据库用于对存储数据进行按需查找和修改。KSQL不进行查找（还），它所做的是连续转换 - 即流处理。例如，假设我有来自用户的点击流以及有关这些用户不断更新的帐户信息表。KSQL允许我对这个点击流和用户表进行建模，并将两者结合在一起。即使这两件事中的一件是无限的。

因此，KSQL运行的是连续查询 - 在新数据通过时连续运行的转换 - 在Kafka主题中的数据流上。相反，对关系数据库的查询是一次性查询 -在数据集上运行一次完成-就像在数据库中有限行上的SELECT语句中一样。

什么是KSQL好？

太棒了，所以你可以不断查询无限的数据流。这有什么用？

1.实时监控满足实时分析

CREATE TABLE error_counts AS
SELECT error_code，count（*）FROM monitoring_stream
WINDOW TUMBLING（SIZE 1 MINUTE）
WHERE type ='ERROR'

其中一个用途是定义实时计算的自定义业务级度量标准，并且可以监视和提醒，就像执行CPU负载一样。另一个用途是在KSQL中为您的应用程序定义一个正确性概念，并检查它是否在生产中运行时满足此要求。通常，当我们考虑监控时，我们会考虑跟踪低级别性能统计数据的计数器和仪表。这些类型的仪表通常可以告诉您CPU负载很高，但它们无法真正告诉您应用程序是否正在执行它应该执行的操作。KSQL允许从应用程序生成的原始事件流中定义自定义度量标准，无论它们是记录事件，数据库更新还是任何其他类型。

例如，Web应用程序可能需要检查每次新客户注册发送欢迎电子邮件时，都会创建新用户记录，并对其信用卡进行结算。这些功能可能分布在不同的服务或应用程序上，您可能希望监视某些SLA中每个新客户发生的每件事情，例如30秒。

2.安全和异常检测

CREATE TABLE possible_fraud AS
SELECT card_number，count（*）
FROM authorization_attempts
WINDOW TUMBLING（SIZE 5 SECONDS）
GROUP by card_number
HAVING count（*）> 3;

您在上面的演示中看到了一个简单的版本：KSQL查询将事件流转换为使用Kafka-Elastic连接器泵入Elastic并在Grafana UI中可视化的数字时间序列聚合。安全用例通常看起来很像监控和分析。您不是在监控应用程序行为或业务行为，而是在寻找欺诈，滥用，垃圾邮件，入侵或其他不良行为的模式。KSQL提供了一种简单，复杂，实时的方法来定义这些模式并查询实时流。

3.在线数据集成

CREATE STREAM vip_users AS
SELECT userid，page，action
FROM clickstream c
LEFT JOIN users u ON c.userid = u.user_id
WHERE u.level ='Platinum';

在公司中完成的大部分数据处理都属于数据丰富领域：从多个数据库中获取数据，对其进行转换，将其连接在一起，并将其存储到键值存储，搜索索引，缓存或其他数据服务中系统。很长一段时间，ETL - Extract，Transform和Load - 用于数据集成是作为定期批处理作业执行的。例如，实时转储原始数据，然后每隔几小时转换一次以实现高效查询。对于许多用例，这种延迟是不可接受的。KSQL与Kafka连接器一起使用时，可以实现从批量数据集成到在线数据集成的转变。您可以使用流表连接来使用存储在表中的元数据来丰富数据流，或者在将流加载到另一个系统之前对PII（个人可识别信息）数据进行简单过滤。

4.应用程序开发

许多应用程序将输入流转换为输出流。例如，负责重新排序在线商店库存不足的产品的流程可能会提供销售和发货流，以计算要下达的订单流。

对于用Java编写的更复杂的应用程序，Kafka的本地流API可能就是这样。但是对于简单的应用程序，或者对Java编程不感兴趣的团队，可能是他们正在寻找的简单SQL接口。

KSQL中的核心抽象

KSQL在内部使用Kafka的Streams API，它们共享相同的核心抽象，用于Kafka上的流处理。KSQL中有两个核心抽象映射到Kafka Streams中的两个核心抽象，允许您操作Kafka主题：

1.流：流是结构化数据的无界序列（“事实”）。例如，我们可以进行一系列金融交易，例如“Alice向Bob发送了100美元，然后Charlie向Bob发送了50美元”。流中的事实是不可变的，这意味着可以将新事实插入到流中，但是现有事实永远不会被更新或删除。可以从Kafka主题创建流，也可以从现有流和表派生流。

CREATE STREAM pageviews (viewtime BIGINT, userid VARCHAR, pageid VARCHAR)
WITH (kafka_topic='pageviews', value_format=’JSON’);

2.表：表是STREAM或另一个TABLE的视图，表示不断变化的事实的集合。例如，我们可以有一个表格，其中包含最新的财务信息，例如“Bob的当前账户余额为150美元”。它相当于传统的数据库表，但通过流式语义（如窗口）进行了丰富。表中的事实是可变的，这意味着可以将新事实插入表中，并且可以更新或删除现有事实。可以从Kafka主题创建表，也可以从现有流和表派生表。

CREATE TABLE users (registertime BIGINT, gender VARCHAR, regionid VARCHAR, userid  VARCHAR)
WITH (kafka_topic='users', value_format='DELIMITED');

KSQL简化了流应用程序，因为它完全集成了表和流的概念，允许使用表示当前正在发生的事件的流来连接表示世界当前状态的表。Apache Kafka中的主题可以表示为KRE中的STREAM或TABLE，具体取决于对主题的处理的预期语义。例如，如果要将主题中的数据作为一系列独立值读取，则可以使用CREATE STREAM。这种流的示例是捕获页面视图事件的主题，其中每个页面视图事件是不相关的并且独立于另一个。另一方面，如果您希望将主题中的数据读取为可更新值的不断发展的集合，则可以使用CREATE TABLE。应该在KSQL中作为TABLE读取的主题示例是捕获用户元数据的示例，其中每个事件代表特定用户ID的最新元数据，无论是用户的名称，

KSQL in Action：实时点击流分析和异常检测

让我们来看一个真正的演示。该演示展示了如何使用KSQL进行实时监控，异常检测和警报。点击流数据的实时日志分析可以采用多种形式。在此示例中，我们标记了在Web服务器上占用过多带宽的恶意用户会话。监控恶意用户会话是会话化的众多应用之一。但从广义上讲，会话是用户行为分析的基石。将用户和事件与特定会话标识关联后，您可以构建多种类型的分析，从简单的指标（如访问次数）到更复杂的指标（如客户转化渠道和事件流）。我们通过展示如何在由Elastic支持的Grafana仪表板上实时连续可视化KSQL查询的输出来结束本演示。

您也可以按照我们的说明自行逐步完成演示并查看代码。

看看里面

有一个KSQL服务器进程执行查询。一组KSQL进程作为集群运行。您可以通过启动更多KSQL服务器实例来动态添加更多处理容量。这些实例是容错的：如果一个失败，其他实例将接管其工作。使用交互式KSQL命令行客户端启动查询，该客户端通过REST API将命令发送到集群。命令行允许您检查可用的流和表，发出新查询，检查状态和终止运行查询。内部KSQL使用Kafka的Streams API构建; 它继承了它的弹性可伸缩性，高级状态管理和容错能力，并支持Kafka最近推出的一次处理语义。KSQL服务器嵌入了这个并添加了一个分布式SQL引擎（包括一些奇特的东西，如用于查询性能的自动字节代码生成）和用于查询和控制的REST API。

Kafka + KSQL 将数据库内部翻出来

我们过去曾谈到过将数据库内部化，现在我们通过在内向外数据库中添加一个SQL层来实现它。

在关系数据库中，表是核心抽象，日志是实现细节。在以事件为中心的世界中，数据库被彻底改变，核心抽象不是表格; 这是日志。这些表仅从日志中派生出来，并在新数据到达日志时不断更新。中央日志是Kafka，KSQL是一个引擎，允许您创建所需的物化视图并将其表示为不断更新的表。然后，您可以针对此类流表运行时间点查询（即将推出KSQL），以持续的方式获取日志中每个键的最新值。

将数据库内部翻出来Kafka和KSQL对公司中所有数据的影响很大，这些数据可以自然地以流式方式表示和处理。Kafka日志是流数据的核心存储抽象，允许进入离线数据仓库的相同数据现在可用于流处理。其他所有内容都是日志的流媒体物化视图，无论是各种数据库，搜索索引还是公司的其他数据服务系统。创建这些派生视图所需的所有数据丰富和ETL现在可以使用KSQL以流方式完成。监控，安全性，异常和威胁检测，分析以及对故障的响应可以实时完成，而不是为时已晚。

KSQL的下一步是什么？

我们正在发布KSQL作为开发人员预览，以开始围绕它构建社区并收集反馈。我们计划在与社区合作时添加更多功能，将其从KSQL的质量，稳定性和可操作性转变为生产就绪系统，以支持更丰富的SQL语法，包括进一步的聚合功能和连续的时间点SELECT表 - 即，除了从流中连续计算结果的当前功能之外，还能够快速查找到目前为止计算的内容。

我如何获得KSQL？

您可以通过使用KSQL快速入门和上述演示来解决问题。我们很乐意听到任何你觉得丢失或方式可以改进：帮腔在＃KSQL通道上铺满社区松弛有任何意见或反馈，并提交GitHub上的问题，如果你发现一个bug; 我们喜欢与早期采用者紧密合作，所以不要害羞。我们期待与其他开源社区合作，将KSQL发展成为一个非常棒的东西。

观看我们在线讲座的录音，了解如何使用KSQL构建实时流媒体应用程序。

最后，如果您对流处理感兴趣并希望帮助构建KSQL，那么Confluent正在招聘

如果你对KSQL可以做什么感兴趣：

观看KSQL的介绍性视频
查看KSQL配方，了解您可以遵循的方便教程和示例
了解 Streaming Audio播客上的KSQL用例

翻译：https://www.confluent.io/blog/ksql-streaming-sql-for-apache-kafka/