本文介绍了我们在Kubernetes中扩展分布式流处理器的经验。流处理器应该支持维护最佳的并行性。然而，添加更多的资源会带来额外的成本，但却不能保证性能的提升。相反，流处理器应该识别出资源需求级别，并进行相应的扩展。

关键要点

流式处理器应该是可扩展的，用以满足流数据处理不断增长的业务需求。
在容器化环境中扩展流式处理器必须在服务质量与相关成本之间做出权衡。
流式处理器应该能够通过水平扩展在容器化环境（例如运行在云服务供应商上的Kubernetes）中利用这种权衡。
在容器化环境中成功运行流式处理应用程序取决于为每个流式处理器配置的硬件资源。
为容器化环境添加更多的硬件资源并不会带来性能的提升。

由于对事件流式处理（即流式处理）应用程序的需求在不断增长，数据流式处理近来已成为数据分析领域的主要范式之一。各种流式处理应用已经出现在各个行业中，如电信、交通流量管理、人群管理、健康信息学、网络安全、金融，等等。

流式处理器是一种软件平台，让用户能够更快地处理和响应传入的数据流。市场上有很多流式处理器可供选择，比如Flink、Heron、Kafka、Samza、Spark Streaming、Storm和WSO2 Stream Processor，它们都是开源流式处理器。

流式处理器的实时操作为提升系统性能提供了高质量的服务。大多数现代流式处理器只需要少数几个计算节点就可以处理90％的流式场景。但是，随着业务的扩展，大多数有利可图的企业需要处理越来越多的工作负载。因此，在选择流式处理器时，必须选择容易扩展且能够处理更大工作负载的解决方案。

流式处理器越来越多地被部署成云端的软件即服务（SaaS），例如Amazon Kinesis Data Analytics、Microsoft Azure Stream Analytics、Google Cloud Dataflow，等等。基于云的流式处理器为在其上运行的流式处理应用程序提供了弹性扩展能力。以容器为中心的管理环境（即容器化环境，例如Kubernetes）可以以可伸缩的方式运行流式处理应用程序。然而，由于硬件或软件环境的异构性和数据流的特性，在容器化环境中何时以及如何进行分布式流式处理器的缩放就成了一个非常重要的问题。本文介绍了一个数据密集型流式处理应用程序的实际应用场景，解释了如何在Kubernetes中进行系统的扩展，以及相关的权衡。我们使用WSO2 Stream Processor（WSO2 SP）作为示例流式处理器，因为它是一个开源的适合实现此类应用程序的云原生流式处理器。不过，我们认为相同的概念也同样适用于市场上其他云原生流式处理器。

我们将提供一个真实的流式处理示例，与检测恶意攻击有关，在这个示例中，有人试图进行未授权的Web服务器登录，并造成服务器拒绝服务（DoS攻击）。一旦检测到此类DoS攻击，系统就会向系统管理员发送警报，以便采取必要的安全措施。

使用Web服务器日志检测恶意攻击

可以使用Web服务器日志中捕获的HTTP日志事件来监控Web服务器的运行状态。如果连续出现相同的HTTP状态码（例如401未授权或403禁止访问），说明有人在尝试恶意登录Web服务器。401响应说明外部第三方尝试访问Web服务器的凭证已经被拒绝。状态码403说明服务器拒绝接受请求，尽管服务器知道如何处理该请求。

图1：检测对Web服务器的恶意攻击

有几种不同的方法可用来处理这种情况。但是，对于这种情况，信息安全专家更喜欢收到实时警报。如果这种恶意请求在三秒钟内超过30次，并且访问率（(未授权请求次数+禁止访问计数)/总请求次数）为1.0，则需要抛出警报。我们使用流式处理器开发了一个警报生成程序（如图1所示），这个处理器接收并处理来自Web服务器的日志事件。为了让系统具备可扩展性，它被部署在运行在Google Compute Engine（GCE）上的Kubernetes集群中。清单1显示了用Siddhi查询语言编写的流式SQL代码。我们将它称为Siddhi应用程序。

清单1：使用Siddhi流式SQL开发的恶意攻击检测程序

@App:name("MaliciousAttacksDetection")

@App:description("HTTP Log Processor for detecting malicious DoS attacks")

@source(type = 'kafka', partition.no.list='0', threading.option='single.thread', group.id="group", bootstrap.servers='kafka-service:9092', topic.list = 'attackDetectionTopic',

        @map(type = 'json'))

define stream inputStream ( iij_timestamp long, ip string, timestamp long, zone float, cik double, accession string, doc string, code float, size double, idx float, norefer float, noagent float, find float, crawler float, groupID int, browser string);

--Output of query 1: I want to know the IP of the malicious hosts which try to make unauthorized login attempts in short intervals.

@sink(type='log')

define stream outputStreamDoSAlert(ip string ,groupID int);

--The Actual latency of parallel siddhi apps are getting started to measure at this point

@info(name = "Query1")

@dist(execGroup='group11' ,parallel ='1')

from inputStream

select iij_timestamp, ip, timestamp, zone, cik, accession, doc, code, size, idx, norefer, noagent, find, crawler, groupID, browser, convert(time:timestampInMilliseconds(),'long') as injected_iijtimestamp

insert into interimInputStream;

--Query 2: Here all the accesses are either 401 or 403 and they have been done at least 30 times within 3 seconds time period.

@info(name = "Query2")

@dist(execGroup='group3', parallel ='12')

partition with (groupID of interimInputStream)

begin

   from interimInputStream#window.timeBatch(timestamp, 3 sec)

   select  ip, count() as totalAccessCount, sum(ifThenElse(code == 401F, 1, 0)) as unauthorizedCount, sum(ifThenElse(code == 403F, 1, 0)) as forbiddenCount,injected_iijtimestamp as iijtimestamp,groupID,timestamp

   insert into #interimStream3;

   from #interimStream3#throughput:throughput(iijtimestamp,"throughput",3,6,"outputStreamDoSAlert",30)

   select ip, totalAccessCount, (unauthorizedCount + forbiddenCount)/totalAccessCount as accessPercentage ,groupID

   insert into #interimStream5;

   from #interimStream5 [totalAccessCount > 30L and accessPercentage == 1.0]

   select ip ,groupID

   insert into outputStreamDoSAlert;

end;

我们将这个应用程序部署在分布式流式处理器中，如图2的部署架构图所示。每个组件（如Worker-1、Worker-2……）都部署为单个容器和单个Kubernetes pod。表1中列出了每个容器类别及其执行的任务。图2显示了整个系统被部署在六个Kubernetes节点中的方案。

表1：Kubernetes环境中不同类型容器执行的任务

我们将应用程序部署在Google Compute Engine的Kubernetes环境中。对于特定的工作负载P，系统应该提供特定的服务质量（QoS）值Q。我们根据Worker的延迟数量来测量Q的值（延迟是指事件进入Worker和事件退出Worker之间的时间差）。下面列出了集群中部署的每个组件。

Web服务器托管在Node 1上；
生产者组件托管在Node 2上，这个组件负责生成工作负载。它读取Web服务器日志并将它们发布到运行在Node 4上的Kafka实例；
Node 2和Node 4运行流式处理器的两个管理器；
由gcloud自动生成的NFS托管在Node 3上；
Worker运行在Node 5和Node 6上，并负责处理实际的工作负载。它们从Kafka实例读取数据，应用流式处理操作，并将结果写回Kafka实例。

图2：Kubernetes集群的部署架构。

但是，随着时间的推移，Web服务器上的工作负载也会增加，这是大多数Web服务器的典型特征。工作负载可能会从P增加到到2P、4P、……、16P，等等。在这种情况下，运行Web服务器监控系统的企业需要维护Q’（QoS属性的观察值），让它与Q相当。流式处理器应该能够扩展到足以维持预期的QoS级别。请注意，可伸缩性是系统处理不断增加的工作负载的能力。在本文中，我们着重关注负载可伸缩性，即随着流量的增加，系统能够正常运行。

有两种方法可以实现负载可伸缩性：强伸缩和弱伸缩。强伸缩在保持问题规模不变的同时增加处理器的数量，弱伸缩也会增加处理器数量，但保持每个处理器的问题规模不变。在本文中，我们采用了弱伸缩，因为我们遇到的是工作负载增加的情况。

实验

我们使用了一个运行在Google Compute Engine（GCE）上的Kubernetes集群。此外，我们使用的节点配备了2核CPU和7.5GB内存，用来托管pod。每个pod都有一个容器，每个容器中都部署了一个流式处理器（SP）组件。我们使用JDK 1.8、Kafka 2.0.1、WSO2-SP 4.3.0和MYSQL 5.7.4来构建docker镜像。它们都作为容器化的应用程序部署在集群中。每个实验需要40分钟时间，包括10分钟的预热。请注意，我们使用符号x-y-z来表示(节点数)-(Worker数)-(实例数)。

我们使用EDGAR日志文件数据集作为实验数据集，因为它已经包含CSV格式的Web服务器日志数据集。我们使用了EDGAR日志文件数据集log20170325.csv。这个CSV文件中包含了22,146,382个事件，文件总大小约为2.4GB，平均消息大小为144字节，每条记录有15个字段。清单2显示了EDGAR数据集中的头两个记录。

清单2：来自EDGAR日志文件数据集的头两个记录

ip,date,time,zone,cik,accession,extention,code,size,idx,norefer,noagent,find,crawler,browser

100.14.44.eca,2017-03-25,00:00:00,0.0,1031093.0,0001137171-10-000013,-index.htm,200.0,7926.0,1.0,0.0,0.0,10.0,0.0,

我们分别测量了三个级别的性能，即节点级别、容器级别和流式处理器级别。不过，本文得出的结论是基于流式处理器级别的延迟和吞吐量。我们在group3 Siddhi执行组中测量了这些值。我们使用Kubernetes集群进行了六次不同的实验，得出了表2所示的结果。

在表2中，ID对应于唯一的实验标识符。节点数量对应于Kubernetes Worker的总数。实例数量是指Siddhi实例的数量。生产者数据速率（线程数）对应于生产者的数量，这些生产者通过从EDGAR数据文件读取HTTP日志事件来生成流事件。

结果

两个节点、6个Worker、6个实例（2-6-6）导致单个工作负载生产者P的平均延迟为390毫秒。这是最基本的情况。随着工作负载的增加，延迟显著增加，而吞吐量会随之降低。我们将工作负载生成器线程增加到16，用以生成很大的工作负载。场景2显示了这种情况。当我们将生成器线程增加16时，延迟增加了28.2％，吞吐量减少了29％。场景3表明，减少输入数据项中唯一组ID的数量会使事情变得更糟（特别是在吞吐量方面）。这是因为减少唯一组ID的数量会导致应用程序的并行性也减少。

如果我们将每个节点的Worker数量加倍，如场景4所示，我们将每个Worker的内存量减半。与情况1相比，吞吐量降低了三分之二，延迟增加了9倍。因此，Worker的内存使用量对应用程序的延迟起着重要作用。有时候，如果没有提供足够的内存，甚至无法部署Worker。

对于第5种场景，我们将Kubernetes Worker节点的数量增加到3个，总共有12个Worker，从而消除了性能瓶颈。现在添加了另一个节点（Node 7），每个节点有四个流式处理器。每个Worker需要1 GB内存，因此每个节点可以有4到5个Worker。因此，即使我们有3-6-6的系统设置（即三个节点，六个Worker和六个实例），只能获得2-6-6的运行性能。但是，2-12-12和3-12-12的性能特征会不一样。

添加更多的硬件资源也无助于是。我们可以从实验场景6中观察到这一点。虽然与场景5相比，节点数量增加了一倍，但我们得到了与场景5相似的平均延迟。如果为Worker添加更多的Siddhi部分应用程序，场景6的方法可能会有用，但这需要额外的内存。

表2：不同Kubernetes集群的性能

当我们在Siddhi应用程序中使用分区时，Siddhi应用程序被分成若干个Siddhi应用程序，这些应用程序将根据分区属性的唯一值分布来获得工作负载。表2最右边的一列显示了groupID字段使用的属性惟一值的数量。在Siddhi应用程序中，我们使用groupID作为分区属性。因此，使用6个惟一值意味着只有6个部分Siddhi应用程序可以根据Siddhi的哈希映射获得工作负载。我们将惟一groupID的数量增加到12来增加并行性。这意味着流将被定向到12个Siddhi部分应用程序。由于这些原因，并行性增加了，我们在场景1中获得的延迟比场景3中的更好。

总结

可伸缩性是容器环境中的流式处理器面临的一个重大挑战，因为应用程序的工作负载会随着时间的推移而增加。本文分享了在Kubernetes环境中扩展此类分布式流式处理器的经验。为此，流式处理器应该提供某种编程语言或查询结构，以维持最佳的并行级别，而不管应用程序的初始规模如何。随着工作负载的增加，需要提供足够数量的硬件资源，让系统可以保持足够的QoS水平。添加更多资源会产生额外成本，但添加更多的资源并不能保证一定会获得性能的提升。流式处理器需要能够识别出资源需求级别，并扩展到可以保持最佳性能与成本比的级别。

关于作者

Sarangan Janakan是WSO2的实习软件工程师，目前是斯里兰卡莫拉图瓦大学计算机科学与工程系的三年级本科生。他的研究兴趣包括数据流式处理和云计算。

Miyuru Dayarathna是WSO2的高级技术主管。他是一名计算机科学家，在流式计算、图形数据管理和挖掘、云计算、性能工程、信息安全等方面做出了贡献。他还是斯里兰卡莫拉图瓦大学计算机科学与工程系的顾问。他最近在WSO2的研究重点是流式处理器和身份识别服务器。他已经在知名的国际期刊和会议上发表过技术论文。

查看英文原文：https://www.infoq.com/articles/distributed-stream-processor-container

创作场景

在容器化环境中扩展分布式流式处理器