大数据任务处理最佳实践：如何用Serverless实现事件驱动？-张果原创

哦哈，大家下午好，不好意思，刚才出了一些小D，这个工厂能出了一些小小的问题。呃呃。呃，我们现在正式开始。呃。呃，各位开发者，呃，同学们大家下午好，我是本次呃service主题的分享人科技的CEO张果。呃，今天我给大家带来的分享是事件驱动架构的，呃，大数据应用。嗯，本次分享的主题主要有三个关键词，呃，一个是一个是事件驱动，一个是大数据。呃，我们今天的演讲主要分为四个部分，第一个部分首先我先介绍一下我们的行业背景，然后第二部分介绍我们的架构思路，第三部分介绍一个我们最近刚刚结束的一个音视频项目的案例，然后最后一个部分介绍我们围绕着我们的项目需求所构建的开发者工具。

呃，首先先简要的介绍我们自己的行业背景，然后我先简单的介绍一下我们的团队，我的团队是在2016年成立的，最早呢是我自己的经济学课题组，因为我本身是做呃经济学专业，然后我的主要研究方向是互联网平台的机制与算法设计，嗯，比如说类似于像大型的电商平台的搜索算法呀等等，或者是呃推荐算法呀这些。然后呢，我们经常需要对海量的互联网公开数据进行采集和处理，比如说从多个电商平台去获取他们的呃搜索排序的价格，然后用来去构建电商价格数据库，去进行一些呃价格水平，以及呃与之相关的一些呃市场市场市场均衡的一些研究。呃，然后呢，然然后呢，我们在这个过程中，我们就发现，呃，我们需要很庞大的计算量，并且我们需要把呃大量的云资，呃大量的计算资源放在云端，而不能放在本地，因为我们真的在本地运行过我们自己的代码，然后发现每天早上都去开一个机实在是太麻烦了。

嗯，然后呢，在呃，我们当时技术团队的带领下，我们就我们就成为了腾讯最早的一批用户。呃，然后在2018年，我们把我们的技术的技术上的积累做成了呃，一家公司，注册了一家公司，然后在呃最近两年，大概在二零年底的时候，我们开始进行系统进行商业化，然后在同时我们进行了原生和改造。呃，我们的主要的业务场景呢，主要是围绕呃，类似于呃大家所比较常见的商业数据分析的呃，一些应用场景，比如说去做一些呃业务数据的呃统计分析，在这个基础之上做一些呃。在做一些呃统计模型，嗯，去去得到我想要的一些因果关系，比如说我们常见的一种点就是嗯，平台制定的一个政策，这个政策对于整个平台的这个开发者有什么影响等等之类的。

呃，我们在这个方向上持持续去投入的主要原因呢，是因为我们认为呃如果想要去彻底的系统的去解决商业数据分析的各种总统的复杂问题，为呃像我们这样的企业以及更大规模的企业去提供更加呃优质的，更加呃客观的，更加呃有性价比的商业商业决策能力。呃，我们需要已知配套的能力来去解决这些问题，并且呢，如果我们能够去呃，把经济学、管理学的研究和商业数据据分析以及商业很好的结在一起的，我们就能为学术界和产业界去创造大的。嗯，所以基于这样的考虑呢，我们发展出了一套，呃，基于原生和的技术体系。

呃，我们在进行呃早期的工作工工作工作的时候我们发现了很多问题，比如说嗯，在传统的虚拟机的模型下，呃扩缩容是非常是非常麻烦的，因此呢，我们对于弹性扩缩容有非常强烈的诉求。嗯，并且呢，由于我们的开发人员主要是呃经济学管理学专业的，就是以这个业务人员业务人员为主，然后我们的重点呢，主要是在于呃呃经济学和管理学的建模，我们希望云的模型能够尽可能的简单，能够最大限度的去用我们手上的各种各样的各呃各种各样现有的一些工具，比如说一些Python代码等等之类的。嗯，同时呢，由于科研项目的对于质量要求，希望我们是比较定望可以足够高的可用性。同时呢，由于我当时是在本科，是在大二的时候开始去做这样的事情，然后呃，本科生在整个学校里面去申请经费是非常困难，所以。

我们我们只有非常有限的呃预算，然后我们希望在这个有限的预算之下，能够去发呃，能够去尽可能的去去去解决这个问题，所以在这种相对比较极的一种场景之下呢，我们就不的会去更加进加更加强大的语。呃，然后这样子的话，我们实际上就会去选择容器集群和函数计算，然后呢，在容器集群函数计算的，我们最终选择了容器集群用来做web，然后用函数计算用来做大数据，主要的一个考虑呢，就是因为函数计算模型本身比较简单，然后呢，它的高可用以及弹性作能力都比较强大。呃，我们下面来简要的去介绍一下函数计算模型的一个计算框架，嗯，函数计算呢，嗯，主要是两个部分组成，首先一个部分是事件个，事件数呢，简理解呢，它就是一个接事件理，它的码通常是原在跑码就可以，比如说它是一个Python的程序，然后呢，我们在外面套一个壳，这个呢叫做件函数，它就是一个入口函数，和很多语言所提供的入口函数其实非常接近。

啊，然后呢，呃，我们我们通过呃，Event event这个参数把用把把来自系统外的进去，然后再把这个event的转发给我们有的程序，这样的话只需要改造一个主函数就可以把原本的程序跑起来。然后另外一个部分呢，是触发器，触发器呢，它主要是用来连接呃云云服务和用户，以及呃云云服务和云服务之间，然后呢，不同的触发器具有不同的功能，比如说API网关呢，它就可以用来去处理HTB的HTB的呃一些嗯请求，这样子的话，用户和人之间，就呃用户和和和和云计算服务以及云计算和云计算服务之间都可以用HTP协议进行进行交互。

嗯，再比如说呢，我们比较常见的在我们的数据，呃，在我们的呃存cos的存在我们的对象存储的存储里面有大量的视频，视频文本数据，然后呢，我们上传了以后，希望立刻去处理它，我们就可用触发器，再比如说我们希望去处理日志中的异常，那我们就可以C触发器。然后呢，这个就是，呃，这个加起来它就构成了函数计算的模型，然后在腾讯呢，主要是腾讯的云函数，这个产品简称是CF。呃，再然后呢，有了这样的一个，呃。在这样的，在这样的计算模型呢，我们有了一个单个的，呃，有了一个单个的计算模型之后，我们还需要去解决函数和函数之间怎么去连接，那么有很多种办办法，呃，其中函数团队提供了一个比较标准的方法呢，就是事件总线。呃，然后事件总线，它的基本原理是从一个云服务，我们叫做事件源去接收一个事件，然后呢，通过一个规则，我们叫做事件规则过滤，过滤完了之后再传递给另外一个用户。

理论上说，它可以对整个云腾讯云体系下的各种各样的服务进行交互，不过呢，我们比较常用的还是使用函数。由于它提供了呃，由于它是基于呃社区的event1.0这个标准协议去做的，所以呃它的格式是很标准的，甚至如果你需要去进行厂商之间的协同，理论上说都是可以的。然后呢，另外呃另外呃另外一个呃重点呢，呃，另外一个可以考虑的选项呢，就是消息队列。嗯，消息队列呢，它也是非常常见的一个，中间它的作用呢，主要就是呃，订阅某一个消息来源，然后再发给另外一个另外一个消息。嗯，消现现在的消息队列呢，嗯，它对于比如说重重啊，死性啊，然后或者对于各种复杂任务的一些调度规则支持都会更加友好，这个呢，对于处理更加复杂的一些，呃，计算需求会非常的有帮助，然后我们在处理各种大数据任务的时候，就经常会发现，呃，会有这样的需求，然后呢，我们会把事件总线，消息队列结合起来。

呃，在这个基础之上呢，我们呃，用用用，用事件，用函数计算和事件总线加起来去做一个，呃我们我们把它称为事件去，呃我们把它称为一个事件流，然后呢，它实际上就是一个典型的事件驱动的架构，然后上一上一个函数是生产者，下一个函数消费者，然后通过事件函数来中转，这样的话，你呃，我们就可以把呃函数和函数串起来，无论有多少个，我们都可以这样串，然后你无论有多少个分支，我都可以通过事件总转发给不同的函数去处理。呃，然后呃，这样子的话，我们就可以用生产者消费者模型或者订阅发布这些比较常见的处理各种各样数据，数据分析应用的方法，可以把原有架构比较理想的移到上。

呃，然后我们下面来讲一个我们自己这边的一个视频项目的一个实例。呃，我们我们前在在一个大型项目中的前期我们找很多研录制了大概，呃，大概有1200个视频，然后这个视频的大小总计是将近三个TB，然后呢，它的大小不一，呃长嗯，有的是录了比较长的时间，有的可能时间比较短，取决于这个呃视频本身的需求，然后它有。呃，几十兆的，然后最大的有十五六个G的，然后平均水平大概在1G左右。我们的主要需求呢，是要把这个视频切块，按照每分钟去切块，然后把每分钟的平均值给计算出来，然后把这个数据最后给到一个CQ里。嗯，这个需求总体来讲听起比较起比较。呃，然后围绕我们刚才做的这种驱动架构呢，我们大概的，嗯，这个云这个方面的一个设计呢，嗯，大概就是围绕我们刚才的架构思路去设计。

嗯，首先是函数计算，就是我们把整个流程分为了三步，第一步呢，首先是从储上去，把这个视频的列表拉下来，然后把它放到呃，放到事件总线里面，这样我们可以分发了。然后呢，我们通过事件总线把它传递给一个下载下载的函数，然后我们通过呃，我们我们我们去cos上去获取这个视频，然后把这个视频给下载到它挂载的文件存储文件里面。然后呢，我再把呃，这个已经下载完的消息再发给下一个函数，然后下一个函数呢，就是用来去做我们刚才所说的这个量平均值的这个计算，然后呢，他获取到了之后，然后从他从从一个CF上去获取这个去获取这个文件，然后呢，他再把这个数据给存到DB里面，就是database里面。呃，然后呃，这样子的话，我们就把存储计算和总线中间给联系在了一起，然后构成了这样的一个项目架。

嗯，这个架构的核心关键呢，或者是说它的重点呢，其实主要是在CS这一。嗯，这一步呢，我们的线线的做法是把呃同一，呃是把同一个项目内的理需求，比如说函数和E函数，这两个函数之间呢，呃就是挂载的cfs使用了同一个。嗯，好处是比较简单，就只需要去递消息就可以，直接我上一个完下一个就可以计算，兼了一个中转站的作用。但它的缺点呢，主要就在于呃，Cfs本身在和呃云进行交互的时候，这个网关本身有颈，然后这个我们这个处理量由于过大，所以会对CF成很大，然后在载过程中们会造成一些异常，以在过程中我们也发现了很多，然后包括在E的传递也会发现很多问题，然后我们会在下一的开发者工具去讲我们具体的一个解决方案。

呃，然后这个项目我们现在目前经过我们打磨过之后呢，这个项目目前的开发，呃，就是开发效率已经到了我们理想的一个就是这样一个T级的规模，我们可以为单位进行敏捷的开发，这个所谓以天为单位的敏捷，指的就是比如说我们今天的工作日，我们开始，呃，把昨天的问题发现或者是上线发现的问题，嗯，总结一下，然后开始排期，开始解决，然后我们可能立刻就会做，然后可能两三个小时就会改，改进完一轮，然后改完了一轮后之后呢，我们可能就会立刻的发布上线。然后会做一些小规模的测试，如果我们觉得没有什么大问题了，我们就直接跑全量，然后这样的话，一轮的开发时间大概，呃，每轮每天大概是就是每天大概是两到三个，就是可能会有个人同时开发，一个人开发以后另一个人接力这种情况出现，然后呢，轮。

注意是全量，不是小规模测试，一轮全量的话，我们大概的测试就是整个单个任务的话，大概可能在一个小时左右，呃，然后整个任务由于弹性容的能力，实际上大概只需要一个半小时到两个小时就能处理，然后这个主要的瓶颈就是刚才所提到E和cfs的一些瓶颈。然后这个是运行时间。然后在成本方面呢，我们目前初步计算的成本就是这样一个全量项目，大概我们全量跑了三四轮，大概了300块钱，然后呃，其中呢，这个存一半，然后计算一半，这样我们其实有计，就是把文件存这个部分给砍掉，所以论说还有30%以上的成本优化空间。嗯，然后整个这样的一个相对复杂的项目呢，代码量总共只有100行，然后测试加这个，这个代码量是含测试的，然后呢，再加上配置文有几十行注释几十行，然起话个项目量大概也就是百样所接近相对小型的微服务，所以非常的好维护。

也是因此呢，所以我们的迭代效率才可以足够的高，然后基本上整个项目从头到尾就只有三个人参与，然后时大概是二十二天，然后平均每天大概两到三个小时，这个工时呢，包括我们我们下面一个部分所要提到轮子的时间，所以实际实际时间的话，大概就实际的工时大概占了不到一半。嗯，然后版本版本数量我们这边发布呢，大概可以达到随时可以发布，然后我们最密集的时候一天发了五个版本，整个项目呢，我发了16个版本，就把它基本上解决了。然后，所以整体的开发率对于像我们这种没有需求复杂团队来就已常理所觉。嗯，但是呢，我们在开发的过程中也发现了很多细节问题，然后呢，围绕这些问题呢，我们把一些业，我们把一些和业务关的一些技术上可复用的一些逻辑给拿出来，然后我们构建了一套呃，重新构建了一套腾讯云的呃SSD。

呃，我们的，呃，我我们我们的主要的推动方式呢，就是一开始首先我们是考虑就是说用代码，然后方便维护，特别是当两个人同时去做的时候，一个人负责去开发业务逻辑，另一个人去负责去做，呃，做基础框架，然后这样子的话，两边就可以同时进行，不会去耽误时间。然后呃，然后呢，在分开迭代的时候，也方便单独进行进行测试。然后呢，我们一开始的做法呢，是在官方的SDK上去进行一些二次装，然后我们很快就发现，因为我们有很多我们自己的Python上的一些实践，我们发现官方的一些实现它的呃，它的实现和我们需求不太符合，另外呢，我们发现我们对于打通各个云产品之间的求非常强烈，你像我们刚才的刚才的一个架构，实际上经已经用到了五个五个云服务了。

啊，然后我们实际上的一个，呃，一个更加大型的，更加复杂的需求，可能会用到更多的。所以我们希望在整体层面上，对于呃官方SDK提供的这种原子化的这种呃这种这种SDK的这种APAPA形式进行一个改进，然后进行一个大幅度大幅度集成。然后所以呢，我们就重新做了一套，然后我们现在已经把它开源了，然后我某会附上附上附上。呃，然后我们的架构呢，主要是分为呃三层，嗯，我们把它定义，我我我们把它这样定义，然后首先是最上层，就是我们直接用这层，我们把它定义为集成a integrity a。这层我们的主要，嗯主主要的作呢，就是呃实现我们特定功能，比如说我们刚才提到的这种下载，下载大文件的时候，发现会有很多不稳定，所以我们在这个A里面加了很多呃断点续传，各种检验的一些能力。

然后呢，下面一层呢，就是API文档中原本规定些样实现。然后最底层呢，就是呃，云厂商的A，腾讯的，我们重新进行些装，然后进行了一些Python，这样方便我们在上层维护的时候会更加方便。然后再进一步呢，实际上我们还同时去做了一些领域模型，然后把呃，比如说像呃对象啊，对象存储里面的对象，然后呃事件总线的云事件，然后比如说云资源这种在整个腾讯云的产品都会用到的一些概念，我们把它建模成一个模型，然后在各个呃各个应用里面再去使用，这样的话我们发现可以大大的去提高它的。呃，整体性，所以所以在我们造完轮过之后，我们整个业业务项目的代码就瞬间就减下来了，所以我们就可以把更多的注意力放在去优化各种各种各种调试上。

然后我们来说一下我们具体的解决方案，然后首先我们发现一个问题呢，是事件投递的效率限制，呃，这个主要有几个原因，一个原因是因为函数本身的，呃，并发扩容是500个分钟，然后我们发现我们一次性都投进去会导致投递过快，然后就会导致在启动的过程中出现出现很多的问题。嗯，然后呃，另一方面呢，就是那个，如果我们投递过快的话，整个下游链。处理上也会有很多压力。嗯，所以呢，我们采取了一个方法，就是我们装了官方投递的这个呃事件事件的API之后，呃，这个它的限制是每是十个，然后呢，我们又分装一个一次性投递的AAPI，然后呢，我们加了一个time参数，直接呃投递一次以后暂停一秒，这样就完的解决了这个投的限制，并且对整个项目的效率也不会有很大的影响。

嗯，当然这个方案并不适合对于实时性要求很高的一些业务，但对于大部分数据处理来说已经绰绰有余了。然后另外一个问题呢，就是我们发现呃在云函数下载的过程中，因为涉及到从cos下载到函数在在在在存到呃文件存储这边，所以这个会有一个比较严重的一个网络性能问题，特别是在你的量过大的时候，超过了整个瓶颈的时候，就会有比较大的网络问题，然后呢，所以我们采取的方法呢，就是在这个下载过程每次都去呃验证，就是每我们首先是把文件分成小文件进行分段下载，然后每次载的时候我们都验证这个小文件有没有下载成功，然后如果下载成功我们就继续下载的，如果下载不成功，我就重新下载一遍。嗯，我们通过这种方法过后之后发现我们由于我们减少了全量的呃，全量的这个需求，所以呢，整个系统的稳定性啊，开发效率啊，运行效率啊，以及成本计费啊都降低了下来，嗯，所以我们发现几个简单的几个简单的A一些实体行的代码，就可以对于整个项目的计费以及效率都会有大大的提高，所以就可以实现我们降本增效的目的。

嗯，然后除此之，除此之外呢，我们因为我们除了用腾讯云的权产品之外，我们在微信系这边也有很多的很多的产品，很多的产品在使用，所以我们做整个腾讯下的其他的SDSDK，目前我们在PPI上已经发已经已经开了企业微信的SDK和coding的SDK。啊，然后目前还有一个微信SDK正在开发之中，然后并且我们还计划把我们前面提到的这个调度逻辑所做成一个呃调度器，然后方便我们自己社区使用。

同时呢，我们也在积极的和云厂商的各个呃产产品团队进行合作，呃目前我们正在推进的一个是我们正在牵头的这边，我们会和呃service产品团队以及开发者呃会合去做一些呃service，呃呃会会做一些service方面的一些项目，然后会做一些社区项及围绕社区做的一些商业化项目。另外呢，就是腾讯微服务的北星团队有在参与今晚开SK。呃，然后呢，我们呃欢迎这个大家和我们去呃合呃进行合作，然后一方面呢，我们是欢迎各个产品线来和我们进行开源和方面合作，我们会我们我们非常愿意帮各个产品线去打磨产品，另一方面就是如果呃企业有呃数据处理的需求，或者是有微服务的需求，或者是有开发开发者工具的需求，都可洽谈。

然后另外呢，就是我们非常欢迎呃，腾讯系的投资人来对我们进行投。和头，我们会计划把我们提到的技术方案进行产品化，然后这个过程需要大量的资金和人员进行参与。啊，然后会后我会把这个我们的呃，我们的呃get的地址放在我们的呃会的群里，然后如果大家想了解我们的话，可以去get up搜索量潮科技，或者去呃微信公众号和呃知乎的机构号去搜索潮科技，然后如果有需要联系我们的话，可以通过群或者是通过我们的官方联系方方式直接联系我们。然后呃，然后呃，我们非常欢迎大家呃来我们进行各种形式的洽谈，我们也在为我们技术去寻找合适的场景。嗯，然后今天的分享就到这里，然后谢谢大家，如果大家有问题的话可以提问。