陈新宇：CKafka在人脸识别PAAS中的应用

本文首发在腾讯云开发者社区，未经许可，不得转载。

我叫陈新宇，在格灵深瞳负责数据流的研发，首先特别感谢如今老师，他们把Kafka一个优秀的消息中间件写出来，也感谢腾讯云做了调优工作，现在就该到我们这些做应用的人用它的时候了，我会从我们应用的层面讲一下它在我们PAAS平台中的应用，讲应用可能很难脱离业务，所以我可能会先给大家解释一下业务，这个业务中的应用，我觉得如何写卡，不卡如何设消费的骨肉普觉得这些东西大家可以自己看看文档，我就不给大家详细的描述了。

自我介绍

我觉得我是一个开源社区的活跃分子，在北京的时候参与了很多社区活动，今天也特别有幸能来深圳参与社区活动，我曾经创办过中国科院的开源镜像站。在15年的时候组织过Apache基金会在中国的路演，现在负责水流的研发，我也有一段在腾讯和IBM的短暂工作经历。

社区活动·北京

这是我的一个副业，现在也是在积极参与各种各样的活动，PPT里左下角这一位，是在阿里纳斯卡上另一个内核的2号人物，葛大神跟他的交流。下面中间这一张是我们组织过的活动里面可能是最受欢迎的一次，大家看到不是因为他有好多小姐姐或者是小妹妹参加，而是因为这个地方是我们IT的一个很神圣的地方——龙泉寺，我不知道有没有人经过，我们跟那边做云计算的法师做交流，可以从云计算一直聊人生的意义，我觉得特别有趣，他们也会举行禅修班的，有兴趣可以关注一下。

所谓计算机视觉

所谓计算机视觉分为几个方向，从处理的东西来讲，可能有图片，有视频；从R识别的方向来讲，有识别人脸和识别人体，以及识别物体，但是能在工业界创造价值的，现在来说基本上是车在安防的场景里边的应用。从运行的平台上讲主要是在gpu上，现在有很多工作也在CPU或者是平台上做。当我们想把这一套东西搬到云上的时候，可能就要考虑问题，比如机器视觉的[云]，有什么不同？视频对于传输的成本是特别的大，把视频再做成深度视频，不知道有没有人了解过深度视频，是商汤或者是face+做深度视频吗？里边会实时的把视频中的车与人，都可以标出来。当我们要在云上做传输时，带宽的成本，存储的成本延迟，还有伸缩性，伸缩性是指GPO到目前为止，可能都没有一个很好的方式做虚拟化，很影响其伸缩性，因为这是一个读者设备，不像CPU可以抢占或虚拟化。所以这些问题就出现了，我们如何在一个云上做一个PAAS平台，我们要做的第一件事情是不做深度视频，而是把视频转化成图片，就会大大降低带宽延迟以及存储成本。第二件事情是把计算的热点尽量分散到CPU和im平台上。接下跟大家解释一下——我们要做一个什么样的平台，它的功能是什么？即我们讨论的问题是数据如何产生价值是应用的本质是什么？数据如何产生价值？

应用的本质

PPT上有三家公司，大家都知道google是怎么赚钱的——收集了很多用户的数据，通过推荐广告盈利；最近这两天闹的沸沸扬扬的Facebook通过手机的敏感信息干扰美国的大选，也是数据产生的价值；头条是怎么赚钱——收集用户的数据做推荐。数据如果单纯的放在一个地方，是不会产生价值的，它一定要跟一个ID产生关联，它才会产生价值。产品经理或者是很多人想方设法收集ID，包括这种账号系统，手机的IMEI，包括wifi热点来收集账号跟数据的关系，我们要做的pos平台也是一样的——不但要识别出脸和脸上的身份，比如说这是一张男生的脸，还是张女士那脸，不用识别它是一个笑着的表情，还是一个不是很开心的表情，而主要是识别出这个人是谁，持续的跟踪或者持续地识别出这是一个人，这个人它属于一个ID的时候才会创造出价值来。这是一个在我们公司内部的demo，我可以拨给大家看一下，第一排倒数第四个是我的那张大脸，家可以看到对人脸的实时记录，当然我们有内部名单会把敏感的人会标出来，现在的数据价值其实不是很大，客户拿了我们对人与ID的识别关系，将其它应用在不同的行业里，因为数据敏感，我没有具体的展示出来，视频我可以跳过，是下面会针对特定人报警，根据ID和行为的关系，可以分析出数据，包括一个中型超市里面的实时数据，进店的人数，男女比例，可能会拿这些数据做分析或者推荐。

深瞳云

所以我们深瞳云主要是用来做什么？是一个认知的计算平台，主要为用户提供ID的对应关系，我们提供的主要是一套数据流，主要的解决的场景是新零售行业、能源行业、社会化安防，还有比如智慧银行在新零售里，我们已经有很多的客户在用。我觉得大家应该能理解，当一个人走到走进店的时候，对他的推荐可以实时地被推出来，就可以根据他的行为做推荐，包括还有汽车赛，根据人的购买能力，做定制化的服务，比如说VIP的识别等等。我们要做的主要是在中间的一层做AI上的推理，或者是比对聚类或实时分析。当事情落实到我身上的时候，其实并不特别的具体化。我们一期的目标是如何接近5万路，即如何把5万路跟刚才我们看到demo接进来，实时推送给客户做分析。当时我们接到这个任务的时候，我们只有四个人的团队，这一个很巨大的任务。

WHY MQ,WHY Kafka

我们想到的第一件事情，就只能依托国有，我们自己做核心的业务。我们接到的任务，是一堆输入，包括存量的抓拍机，或者是我们自己的公司做的机器人产品，和我们之前在安防行业做过的AA推理病情和比率的引擎，识别人脸和比对人脸相似度的的引擎。第一步开始考虑我们为什么需要一个消息队列。因为我们提供的是一个有状态的场景，一个数据流场景，我不知道大家能不能理解，数据从一个摄像头上倒推送到客户，他是有价值的，没有价值的，这是一个流动的过程。它不像face+或者是微软认知服务的APP，可以发一张过并反馈结果，这种一次性的行为可以拿一篇做，但是比如说这个人从A出口进了一个店，在店里面逛了两个小时，从B出口出的时候，要记录它整个的行为，这个过程中是有状态存在的，我们还要对这个状态做实时的跟踪分析，所以必须得用数据流解决而不能用APA。其次我们要有一个高吞吐量的要求，我们一期可能会有5亿张/每天的要求，大概是每秒5000张，我们也会给自己立一个小目标——比如说我们万一做到对吧，5万张每秒，还有比如延迟的要求，为什么用一个消息队列，至于为什么没有Kafka，是下面我们用它解决高吞吐量的问题，解决结果的问题，缓存的问题，更重要的是它有良好的社区和客户端的支持。其实我们公司也是一个以波段为主的公司，我们也进行了大量的测试，都很稳定，没有出过太多的问题。主要是有良好的社区生态和客户端支持，因为我们这套系统其实在是一个PAAS平台，有特殊的客户会拿私有化的部署，或者是不部署到其他的云上，这种公有云的支持也是特别重要，选择本来就不多，好在Kafka已经全满足了需求。

带宽

如果一张图的大小是50K，如果我们每秒发5000张，带宽大B可能是要2000兆，对于我们一个精打细算的团队来说是不能接受，带宽的费用会变得特别高。如果我们要再做的多，费用就会变。公有云上带宽的收费其实不是线性的，带宽越大收费越多，解决方案是把图片直接从前端的上传到对象存储，让对象存储分担流量的压力，通过存储回调触发这个数据流的计算。第二个问题是Kafka的带宽。Kafka的带宽大家基本上可以理解为入口的带宽乘以topic的数量，如果把这些图都塞到Kafka里面做计算，显然是不可接受的。解决办法是用url，再将图片的数据在整个处理过程中尽量把整个消息的大小控制在1K左右，其实腾讯云上Kafka的带宽的要求已经完全满足需求，所以这个任务就是万里长征走出了第一步，我们把设备已经能接进来，扔到Kafka里，后面是处理的问题。

如何降低计算的成本？

第一个方向是用边缘计算分担计算的压力，在arm平台上，包括像声控机器人，其实是安卓的平台，它里面用CPU和im分担计算压力，我们自己做的前端，包括海康的大公司，他们做的前端都支持了对人脸的最基本检测，我可以不识别出来你是男生还是女生，但是我知道你是一张脸，知道你是一张脸，这就足够了，就可以把这张脸从图里抠出来，上传上来的我们就降低了图片传输的成本，也因为我们一个摄像头不可能无时无刻都把图片传上来，那是视频了，对吧？用边缘计算的方案，也会解决运维的问题。第二个问题是利用时空的特性做分级，如果对一个商场里面的一个人做分类或者是做比较，它的运算量肯定是低于在所有的租户或者整个范围内部的。我们就用两种方法用CPO的模型和CPU的模型，它的效果可能没有GPU好，但是在小的范围内识别率还是很高的，比对直接放在应用内部，在内存中进行，一是会少一次调用，二是我们可以把这些全都做成无服务的应用，把它塞到kubernets里，好做负载均衡，比如CPU在一个group范围内，用Global的比对，或者是特征的计算引擎做计算，会降低整个计算的一整个数量级。

如何做实时行为分析？

比如说一个人进了商场，他现在是什么样，在哪里停留得久，停留得少，像这样的信息其实是我们客户很希望能实时得到的。这就有一个问题——如何保证同一组camera的图片分发到相同的处理节点，我们不可能做一个单体把这么大的问题撑起来，肯定是一个分布式的，这就涉及到分发的问题，我们之前尝试过一种简单的方法——利用Kafka的特性，但会造成数据倾斜，不同的场景里面，可能数据产生的量是不一样的，这种解决问题的方法是显然的不够优雅，会有运维的问题。其次，如何保持这个状态的热更新，还有异常的恢复，当节点挂掉的时候，或者是我们做升级的时候，我们已有的这些状态怎么保持下来？最简单的，如果一个人进了一个商场，我们会告诉客户，这个人到你的商场，我们不能因为一次更新，这个人从来都没有出过，这个状态的维护是非常的重要。解决的方法是用流计算的框架Apache Flink。它有助这些特性，我就不一一的念，它自己有时间窗口，最主要的是这种容错功能和流的分发功能，能把问题更好的解决。当然刚才让饶军老师也讲了，Kafka可能将来在流处理上会有其他的升级可能会考虑。经过刚才那么几个问题，是系统可以变成PPT所示的样子，数据通过这个模块，进了Kafka，再在局部上我们特征上做了一次比对，筛选出来，有优良的特征，再拿到Filnk，做人的行为分析，就整个流程基本上能跑完。

结果的入库与实时查询

虽然我们给用户提供了API把流导给用户，他们可以访问，但是数据也要保存下来，供以后查验，如何把数据写到数据库里？后如何提高写入的性能？如何做高可用？我们也不想用这种写入，会严重影响到流处理的高效性，没什么方法解决，因为数据结果已经被推送到客户，被拿用来做各种各样的事情，其实我们写的数据结果也是这些东西，只要做一做对应的解析，写到数据库里就可以了。IT就提供了很方便的方式，包括高可用，包括offset的管理，跟Kafka整合的特别好。offset的管理是还给它本身会提供至少一次的语义投递，要是想自己做，实现恰好一次的这种投递，可能要自己还要做其他的工作，比如说自己管理offset。在内部一条一条的查数据，不能满足我们高大大批量写入的需求。另一个是有一个插件，列在缸中的页面上，但我用了之后发现因为这个项目不是很活跃，重新把它实现并优化了一遍，自己做了一个，当然上面的名字是我们内部的代号，可能会换一个名字，并把开源出来，贡献到社区里，我们一起维护的对象是posS是pg数据库。为什么要做查询？因为实时的数据，当一个人的行为还没有完成的时候，我们没有办法把它写到库里，如果你要写到库里，会很大的影响整个系统的性能，我们把实时的数据进行缓冲，提供一套API给用户查询。查询分两部分，一部分是历史数据，大概的实时数据，大家可以看到上面多了东西，是这个地方是一个消息分发的模块，会对消息做去重。其实在我们看来为了维护的方法，首推这种形式推送给客户的，是因为Kafka实在是太火了，或者是太好用了，我们很多客户强烈要求他们只要Kafka，所以没有办法，所以我们也加了对Kafka推送的支持。当然在公网上可能会有加密的需求。

配置变更

整个过程中其实我们少画一条，在这个地方，其实当数据被上来的时候，我们会要为它加上元数据，是刚才饶军老师说的。当数据配置发生变更的时候，我们怎么应用到数据流上来？这时有两种做法，是左边传统的这种，其实是两次提交，往数据库里写一份，再往你的开始或者是别的地方写一份，就有一致性的问题，或者是实现上的复杂度不够的优雅，后面这种方式是先把数据写到数据库里，监听数据库上发生的数据变更，把变更再写到Kafka，我们其他业务系统就会读Kafka的topic，把它拿出来后应用到不同的系统，比如说上面上午逻辑就只用写一遍，如果不是，在左边的场景里要加入新的数据写入对象，要改整个代码，做到更优雅的实现。但这个场景是比较特殊，因为配置不需要立马发生变更或者是应用，在客户看来可能三秒五秒配置能够得到更新都是OK的，但实际过程并没有这么长的时间。有了这些之后，配置变更下面这条线，监听数据库的变更，把它通过模块，监听并写到Kafka。我们把数据导出来之后，不但是应用到配置上，在实时的行为分析中，没办法做太多历史数据统计或者是分析，只能做实时的。在整个过程中，大家可以看到Kafka起到了很多的连接作用。整个系统是围绕着Kafka构建的，我们用了Kafka做缓冲，解耦，然后做配置变更，往不同的数据对象导数据，甚至把数据最后提供给客户。Kafka在整个过程中是核心。日志的收集，其实是用了Kafka做了一次缓冲，

Kafka使用建议

第一个是按照团队内部具体的要求，再把它做一次封装，只要实现三个方法，初始化工作写到start里，flush的时候，他会应用提交offset，可以统一的管理Kafka的，因为对不是很了解的新手来说，可能很难把按照要求，比如说准确一次的投递或者是至少一次的投递，把程序实现出来。如果你封装成一个包，会很方便他们使用。

第二个是这种消息格式，最好是避免不友好的编码方式，即使有很高的压缩的比例，对以后的检索或者是定位问题是很不方便的。我们还是推荐用Kafka本身的这种消息压缩的机制做消息的压缩。

第三是在数据流的源头创建，唯一的ID在后续的这些处理上都把ID带着，方便查找和定位问题，其实我们自己在log上都分了一个包，。

最后一点是控制消息的大小，尽量缩小在一个可控的1K以内，提高它的性能

结论

我们团队四个人在做这个事情的时候，有一个体会——要相信开源社区的力量，让公有云做他们该做的事情，比如会解决大部分的运维问题，选一个可靠的框架，不要自己造轮子，要不然你的生产力会大大的下降。

最后是感谢Kafka项目，感觉感谢赵军老师，感谢陈云飞卡尔团队，在稳定性很高兴各种方面做的努力，我们提了很多的，特别的感谢。

QA

Q：你好，我想问一下你们那个系统有没有做持久化的？只是做实时吗？

A:持久化是两个方面，第一是我们刚才在前面讲了有一个入库的操作，通过class把数据写到数据库里，post过来的数据。那个是持久化，里面是一个数据仓库，相当于也是一个持久化。假设这个场景之前记录了用户的人脸，对吧？现在我给你一个用户的人也能够立刻找到这个用户。这是我们实际的需求，是完全可以的，但是API是没有数据流量，比对引擎会记下来，每张脸他会在。因为GPU的运算能力很强，会在千万级的数据上秒级的反馈出来，找到跟你最相近的人脸。原理是对人脸的特征提取，按照不同的精度，可能大概有好几百位的浮点数，计算两个向量之间的相似度，其实是在比对两个之间的距离，要在几千万的人里面计算各相似度。

Q：你好，我想问一下Kafka跟MQ之类的，一般传说的消息，包括文字型的，你刚才提到图片是你们另外有一个接口还是什么转换之类的？

A：没有图片，消息是没有放在这个Kafka里边的，因为图片特别的大，我们最小的一张图大概有二十三十K左右，消息的负担会变得特别大，所以我们其实没有画出来，在这个地方提取特征的时候才会把图拉出来。