世界上90%的数据是在过去几年里产生的——事实上，过去30年中，全球的数据量大约每两年增加10倍，这远远超过了计算机领域的摩尔定律。“数据的海量增长向人类提出一项要求，那就是我们必须迅速掌握像人工智能这样的技术手段，来应对大数据提出的挑战。”曙光公司首席运营官叶健如是说。

“大数据平台助力提高分析与推理速度，高性能计算则帮助提高了AI的学习速度。”叶健表示，新信息环境与人工智能技术相结合将催生“实时系统智能”，例如智慧城市、智能医疗、智能交通等。

大数据、高性能计算、人工智能相得益彰

“由于各类传感器和数据采集技术的发展，人们开始拥有以往难以想象的海量数据，同时也开始在各个领域拥有深度的、细致的数据。” 叶健说到，这些正是训练该领域“智能”的前提。大数据与AI发展都需要“快”，而“快”恰是曙光的特点和基因。

叶健说，海量数据、高速并行运算、更优化的算法共同促成了人工智能的突破性发展；而互联网环境中的许多环节都是在一边学习一边应用的过程中实现的。大数据、高性能计算、人工智能三者像是踏上了一个共同前进的节拍，相得益彰。

“与以往传统的算法相比，立足于神经网络的人工智能技术利用输入的数据自行模拟和构建相应的数学模型。”叶健解释说，这一算法特点决定了它更为灵活且可以根据不同的训练数据而拥有自优化的能力。

如今，人工智能在大数据及高性能计算的驱动下，已经在各个领域实现了价值。商品推荐、社交图谱、智能客服乃至无人驾驶等都已经依靠深度学习和其他人工智能方法，实现了巨大的商业价值。

助力社会走向实时系统智能

纵观人工智能60多年的发展史和历史上的几次起落，人工智能这次的变革“大不同”。叶健分析说：“首先，人工智能的基础发生了改变，已经从以往的样本分析变成了如今的大数据、多媒体数据、传感器网的流数据以及AR和VR的复杂系统分析。其次，从社会需求来看，人工智能不再满足于研究个体的智能，而是希望研究整个系统的智能，甚至是实时系统智能。”

“随着虚拟世界与物理世界的界线不断模糊甚至趋近于相互融合，人工智能正在走向2.0时代。” 叶健表示，实时系统智能已在诸多应用场景中大显身手，如通过预研人脸采集比对预警、实时视频图像预警、视频录像情报分析系统预研防控及情报业务，启动情报加指挥双引擎，达到反恐的实时布控；通过利用入侵检测系统提供的功能并结合异常分析，准确报告攻击行为，实现网络犯罪模式的实时识别；移动警务视频云平台通过支持数千点位移动警务终端的视频直播和点播等功能，做到视频监管立体全覆盖，以利于快速研判，为重大活动道路交通疏导、交通事故现场处理、交通应急事件处置等提供有效决策依据和业务支持。

叶健认为，未来，新一代人工智能应该沿大数据智能、群体智能、跨媒体智能、人机混合增强智能和自主智能系统方向发展，并在商务、医疗、制造等各领域产生深刻影响。

“曙光2015年提出‘数据中国’战略，目前已经积累了大量的云计算和大数据技术与应用经验，接下来，曙光将借助人工智能技术优势，在‘数据中国’的基础上进一步实施‘数据中国智能计划’，致力于将人工智能和中国的信息化合为一体，使信息化的线路从数字化发展到网络化，最后进展到智能化。”叶健说，在人工智能变革大潮的机遇中，曙光将依靠大量数据的滚动，将数据的累积、迭代和自动标注形成一个良性循环。

“本文作者 陈睿 ，优知学院创始人新技术、新趋势往往趋之若鹜却又很难说的透彻，希望这篇文章能让大家完整的理解什么是大数据:该篇包含：大数据、技术、场景应用以及大数据的岗位。

什么是大数据

说起大数据，估计大家都觉得只听过概念，但是具体是什么东西，怎么定义，没有一个标准的东西，因为在我们的印象中好像很多公司都叫大数据公司，业务形态则有几百种，感觉不是很好理解，所以我建议还是从字面上来理解大数据，在维克托迈尔-舍恩伯格及肯尼斯库克耶编写的《大数据时代》提到了大数据的4个特征：

一个是数量大一个是价值大一个是速度快一个是多样性第一个是数量比较大，只有数据体量达到了PB级别以上，才能被称为大数据。1PB等于1024TB，1TB等于1024G，那么1PB等于1024*1024个G的数据。

第二个是价值大，你如果有1PB以上的全国所有20-35年轻人的上网数据的时候，那么它自然就有了商业价值，比如通过分析这些数据，我们就知道这些人的爱好，进而指导产品的发展方向等等。如果有了全国几百万病人的数据，根据这些数据进行分析就能预测疾病的发生，这些都是大数据的价值。

第三个就是多样性，如果只有单一的数据，那么这些数据就没有了价值，比如只有单一的个人数据，或者单一的用户提交数据，这些数据还不能称为大数据，所以说大数据还需要是多样性的，比如当前的上网用户中，年龄，学历，爱好，性格等等每个人的特征都不一样，这个也就是大数据的多样性，当然了如果扩展到全国，那么数据的多样性会更强，每个地区，每个时间段，都会存在各种各样的数据多样性。

第四个是速度快，就是通过算法对数据的逻辑处理速度非常快，1秒定律，可从各种类型的数据中快速获得高价值的信息，这一点也是和传统的数据挖掘技术有着本质的不同。

大数据的行业应用

大数据无处不在，大数据应用于各个行业，包括金融、汽车、餐饮、电信、能源、体能和娱乐等在内的社会各行各业都已经融入了大数据的印迹。

制造业，利用工业大数据提升制造业水平，包括产品故障诊断与预测、分析工艺流程、改进生产工艺，优化生产过程能耗、工业供应链分析与优化、生产计划与排程。

金融行业，大数据在高频交易、社交情绪分析和信贷风险分析三大金融创新领域发挥重大作用。

汽车行业，利用大数据和物联网技术的无人驾驶汽车，在不远的未来将走入我们的日常生活。

互联网行业，借助于大数据技术，可以分析客户行为，进行商品推荐和针对性广告投放。

电信行业，利用大数据技术实现客户离网分析，及时掌握客户离网倾向，出台客户挽留措施。

能源行业，随着智能电网的发展，电力公司可以掌握海量的用户用电信息，利用大数据技术分析用户用电模式，可以改进电网运行，合理设计电力需求响应系统，确保电网运行安全。

物流行业，利用大数据优化物流网络，提高物流效率，降低物流成本。

城市管理，可以利用大数据实现智能交通、环保监测、城市规划和智能安防。

生物医学，大数据可以帮助我们实现流行病预测、智慧医疗、健康管理，同时还可以帮助我们解读DNA,了解更多的生命奥秘。

体育娱乐，大数据可以帮助我们训练球队，决定投拍哪种题财的影视作品，以及预测比赛结果。

安全领域，政府可以利用大数据技术构建起强大的国家安全保障体系，企业可以利用大数据抵御网络攻击，警察可以借助大数据来预防犯罪。

个人生活， 大数据还可以应用于个人生活，利用与每个人相关联的“个人大数据”，分析个人生活行为习惯，为其提供更加周到的个性化服务。

大数据的价值，远远不止于此，大数据对各行各业的渗透，大大推动了社会生产和生活，未来必将产生重大而深远的影响。

大数据使用的技术

说起大数据，大数据有三个层数据采集、存储、计算三层。

第一个是数据采集层，以App、saas为代表的服务。

大数据基础阶段需掌握的技术有：Linux、Docker、KVM、MySQL基础、Oracle基础、MongoDB、redis以及hadoopmapreduce hdfs yarn等。

第二个数据存储层，比如云存储，需掌握的技术有：hbase、hive、sqoop等。

比如：Hadoop作为一个开源的框架，专为离线和大规模数据分析而设计，HDFS作为其核心的存储引擎，已被广泛用于数据存储。HBase，是一个分布式的、面向列的开源数据库，可以认为是hdfs的封装，本质是数据存储、NoSQL数据库。

HBase是一种Key/Value系统，部署在hdfs上，克服了hdfs在随机读写这个方面的缺点，与hadoop一样，Hbase目标主要依靠横向扩展，通过不断增加廉价的商用服务器，来增加计算和存储能力。

第三个是数据计算应用层，以数据为基础，为将来的移动社交、交通、教育，金融进行服务，涉及到大数据架构设计阶段需掌握的技术有：Flume分布式、Zookeeper、Kafka等，以及大数据实时计算阶段需掌握的技术有：Mahout、Spark、storm。

大数据的岗位

1.数据分析师Data analyst

指熟悉相关业务，熟练搭建数据分析框架，掌握和使用相关的分析常用工具和基本的分析方法，进行数据搜集、整理、分析，针对数据分析结论给管理销售运营提供指导意义的分析意见。

2.数据架构师Data architect

对Hadoop解决方案的整个生命周期进行引导，包括需求分析，平台选择，技术架构设计，应用设计和开发，测试和部署。深入掌握如何编写MapReduce的作业及作业流的管理完成对数据的计算，并能够使用Hadoop提供的通用算法， 熟练掌握Hadoop整个生态系统的组件如： Yarn，HBase、Hive、Pig等重要组件，能够实现对平台监控、辅助运维系统的开发。

3.大数据工程师Big DataEngineer

收集和处理大规模的原始数据（包括脚本编写，网页获取，调用APIs，编写SQL查询等）；将非结构化数据处理成适合分析的一种形式，然后进行分析；根据所需要的和专案分析商业决策。

4.数据仓库管理员

Data warehousemanager：指定并实施信息管理策略；协调和管理的信息管理解决方案；多个项目的范围，计划和优先顺序安排；管理仓库的各个方面，比如数据外包，移动，质量，设计和实施。

5.数据库管理员Database manager

提高数据库工具和服务的有效性；确保所有的数据符合法律规定；确保信息得到保护和备份；做定期报告；监控数据库性能；改善使用的技术；建立新的数据库；检测数据录入程序；故障排除。

6.商业智能分析员Businessintelligence analyst

就工具，报告或者元数据增强来进行传播信息；进行或协调测试，以确保情报的定义与需求相一致；使用商业智能工具来识别或监测现有和潜在的客户；综合目前的商业只能和趋势数据，来支持采取行动的建议；维护或更新的商业智能工具，数据库，仪表板，系统或方法；及时的管理用户流量的商业情报。

作者：Jon Bratseth

从2006年开源Hadoop开始，Yahoo（也就是现在的Oath）为广大开发者社区的大数据基础设施做出了不可磨灭的贡献。现在，我们又卖出了坚实的一步，Yahoo的大数据处理和服务引擎Vespa正式在GitHub上开源了（https://github/vespa-engine）。

越来越多的应用程序需要处理大量的数据，尽管开发者可以使用Hadoop来存储和批处理数据，也可以使用Storm来处理流式数据，但这些技术无法直接服务于最终用户。提供大规模服务是一项巨大的挑战，当用户需要等待基于大量数据集的计算结果时，比如特征搜索、推荐系统、定制化，这种挑战就会变得尤为明显。

有了Vespa，开发者可以轻松地构建基于大数据集实时计算结果的应用，而到目前为止，只有少数几个大公司具备这样的能力。

提供服务不是简单地根据ID查询项目或应用模型计算出几个数字，很多提供服务的应用需要基于大数据集运行计算，比如搜索和推荐。为了给用户返回精确的搜索结果或推荐清单，应用程序需要找出所有符合查询条件的项目，根据相关度或推荐模型决定每一个项目的匹配程度，移除重复项目，增加浏览辅助项，最后把结果返回给用户。因为这些计算依赖用户的具体请求，所以无法预先计算。应用程序必须实时地处理用户请求，而且要快，因为用户在等待结果。在大数据集上快速执行上述操作需要很多基础设施的支持——分布式算法、数据分布和管理、高效的数据结构和内存管理等等。而这些正是Vespa要为开发者提供的——一个一站式的引擎，简洁易用。

我们已经在Oath的多个产品上使用了Vespa，包括Yahoo、Yahoo News、Yahoo Sports、Yahoo Finance、Yahoo Gemini、Flickr等。我们使用Vespa来处理每日数十亿个用户请求，为用户返回基于数十亿文档得出的搜索结果和推荐内容，并提供定制化内容和广告。事实上，Vespa每秒可以处理90,000次内容和广告请求，延迟通常只有几十毫秒。在Flickr上，Vespa每秒可以处理数百个基于数百亿张图片的关键字和图像搜索。另外，Vespa在Yahoo Gemini上每天处理30亿个请求，高峰时段每秒钟140,000个请求，直接给公司带来可观的收益。

我们基于Vespa构建的应用具备如下特性：

使用SQL风格的查询和文本搜索来过滤内容

基于匹配项生成数据驱动的页面

根据人工或机器学习相关度模型对匹配进行排名

毫秒级的处理响应时间

实时写入数据，每个节点每秒钟写入数千次

在运行服务的同时进行伸缩和重配置

为了保证速度和伸缩性，Vespa在多台机器上分布数据和计算任务，避免了单点master的瓶颈。传统的应用将数据拉取到一个无状态的层上进行处理，而Vespa是将计算任务推送给数据集。为此，Vespa需要做很多非常棘手的工作，比如当机器发生故障或增加新机器时在后台重新分布数据、实现分布式的低延迟和处理算法、处理分布式数据一致性等。

我们在构建alltheweb（后被Yahoo收购）时就开始开发我们的搜索和服务。在过去几年中，我们不断使用新技术重写了我们的引擎。Vespa是我们发布过的项目当中涉及范围最广、代码量最大的一个项目。Vespa已经在Yahoo的大部分关键系统上得到了实地验证，所以我们很高兴能够把Vespa推向世界。

Vespa赋予了开发者将任意大小数据集和模型填充进服务系统的能力，而且可以实时地得到计算结果，带来更好的用户体验，而且成本更低，比预计算方式具有更低的复杂度。而且，开发者可以与复杂的计算展开实时的交互，不需要启动离线作业或反复回过头来检查结果。

Vespa可以运行在自有数据中心或云端。我们提供了Vespa的Docker镜像和rpm安装包，也提供了运行指南，可以让Vespa运行在本地机器或AWS集群上。

这个（http://docs.vespa.ai/）是我们的文档，里面包含了一个入门指南。

管理分布式系统不是件简单的事情。我们投入了大量精力开发Vespa，其他开发者就可以专注在创建功能上，他们可以实现基于大数据集的实时计算，而不是把时间花在集群和数据的管理上。根据我们的文档所给出的指南，你可以在不到十分钟的时间内让一个应用跑起来。

End.

什么是Kafka？

Apache Kafka是一个开源项目，提供强大的连续数据流分布式处理 - 目前全球已有数千家企业信赖其产品，包括Netflix，Twitter，Spotify，Uber等。

技术，体系结构和实施使其高度可靠和高度可用，使流处理应用程序能够利用地理分布的数据流

实时数据流

通过实时数据，您可以实时处理并做出反应。

Kafka实时数据流可帮助您在事件发生时及时获得反馈，并将其用于您的优势。历史数据，加上来自Kafka的实时数据流，可帮助您为未来做出重要决策。实时数据可帮助您获得竞争优势，并使您可以更有效地使用大数据。有效的实时数据流是大多数现代应用程序的核心，架构设计不再只是一个数据队列。

像Instaclustr这样的托管服务提供商在其数据平台上提供Apache Kafka将提供全天候的专家支持，并持续监控吞吐量，延迟时间并全天候进行必要的调整，以便您继续享受Apache Kafka的优势

为什么超过三分之一的财富500强公司使用Kafka

Kafka同时支持写入和读取可扩展性。这意味着您可以将大量数据传输到Kafka，并同时执行消息的实时处理，包括将消息发送到其他系统。

这些应用程序实际上仅限于您的想象力。

哪些行业已经在使用Kafka？

kafka正在各个行业使用，包括物流，零售，医疗保健，金融服务，电子商务，物联网等。

例如在物流行业，Kafka正在帮助更快地移动软件包并帮助公司实现盈利。考虑到物流的真实世界复杂性，最好尝试跟踪货物，仓库和卡车的位置。当与这3个参数相关的实时数据通过Kafka管道时，可以收集有助于各种不同方面的信息，例如收集，存储，交付，计划和优化货物移动，实时检查，审计和欺诈检测。

同样，医疗保健行业中的患者医疗记录和医疗测试也是保险供应商的要求，以及设施管理，床位管理和患者EMR。Kafka管道有助于处理不同的情况。

文 | 潘国庆 携程大数据平台实时计算平台负责人

本文主要从携程大数据平台概况、架构设计及实现、在实现当中踩坑及填坑的过程、实时计算领域详细的应用场景，以及未来规划五个方面阐述携程实时计算平台架构与实践，希望对需要构建实时数据平台的公司和同学有所借鉴。

一、携程大数据平台之总体架构

携程大数据平台结构分为三层：

应用层：开发平台Zeus（分为调度系统、Datax数据传输系统、主数据系统、数据质量系统）、查询平台（ArtNova报表系统、Adhoc查询）、机器学习（基于tensorflow、spark等开源框架进行开发；GPU云平台基于K8S实现）、实时计算平台Muise；

中间层：基于开源的大数据基础架构，分为分布式存储和计算框架、实时计算框架；

离线主要是基于Hadoop、HDFS分布式存储、分布式离线计算基于Hive及Spark、KV存储基于HBase、Presto和Kylin用于Adhoc以及报表系统；

实时计算框架底层是基于Kafka封装的消息队列系统Hermes, Qmq是携程自研的消息队列， Qmq主要用于定单交易系统，确保百分之百不丢失数据而打造的消息队列。

底层：资源监控与运维监控，分为自动化运维系统、大数据框架设施监控、大数据业务监控。

二、架构设计与实现

1.Muise平台介绍

1）Muise是什么

Muise，取自希腊神话的文艺女神缪斯之名，是携程的实时数据分析和处理的平台；Muise平台底层基于消息队列和开源的实时处理系统JStorm、Spark Streaming和Flink，能够支持秒级，甚至是毫秒级延迟的流式数据处理。

2）Muise的功能

数据源：Hermes Kafka/Mysql、Qmq；

数据处理：提供Muise JStorm/Spark/FlinkCore API消费Hermes或Qmq数据，底层使用Jstorm、Spark或实时处理数据，并提供自己封装的API给用户使用。API对接了所有数据源系统，方便用户直接使用；

作业管理：Portal提供对于JStorm、Spark Streaming和Flink作业的管理，包含新建作业，上传jar包以及发布生产等功能；

监控和告警：使用Jstorm、Spark和Flink提供的Metrics框架，支持自定义的metrics；metrics信息中心化管理，接入Ops的监控和告警系统，提供全面的监控和告警支持，帮助用户在第一时间内监控到作业是否发生问题。

2.Muise平台现状

平台现状：

Jstorm 2.1.1、Spark 2.0.1、Flink1.6.0、Kafka 2.0；

集群规模：

13个集群、200+台机器150+Jstorm、50+Yarn、100+ Kafka；

作业规模：

11个业务线、350+Jstorm作业、120+SS/Flink作业；

消息规模：

Topic 1300+、增量 100T+ PD、Avg 200K TPS、Max 900K TPS；

消息延时：

Hermes 200ms以内、Storm 20ms以内；

消息处理成功率：

99.99%。

3.Muise平台演进之路

2015 Q2~2015 Q3 ：基于Storm开发实时计算平台；

2016 Q1~2016 Q2 ：Storm迁移JStorm、引入StreamCQL；

2017 Q1~2017 Q2 ：Spark Streaming调研与接入；

2017 Q3~2018 Q1 ：Flink调研与接入。

4.Muise平台架构

1）Muise平台架构

应用层：Muise Portal 目前主要支持了 Storm 与 Spark Streaming两类作业，支持新建作业、Jar包发布、作业运行与停止等一系列功能；

中间层：对底层Infrastructure做了封装，为用户提供基于Storm、Spark、Flink相对应的API以及各方面Services；

底层：Hermes & Qmq是数据源、Redis、HBase、HDFS、DB等作为外部的数据存储、Graphite、Grafana、ES主要用于监控。

2）Muise实时计算流程

Producer端：用户先申请Kafka的topic，然后将数据实时写到Kafka中；

Muise Portal端：用户基于我们提供的API做开发，开发完以后通过Muise Portal配置、上传和启动作业；作业启动后，jar包会分发到各个对应的集群消费Kafka数据；

存储端：数据在被消费之后可以写回QMQ或Kafka，也可以存储到外部系统Redis、HBase、HDFS/Hive、DB。

5.平台设计 ——易用性

首先：作为一个平台设计第一要点就是要简单易用，我们提供综合的Portal，便于用户自己新建管理它的作业，方便开发实时作业第一时间能够上线；

其次：我们封装了很多Core API，支持多套实时计算框架：

支持HermesKafka/MySQL 、QMQ；集成Jstorm、Spark Streaming、Flink；作业资源管控；提供DB、Redis、HBase和HDFS输出组件；基于内置Metric系统定制多项metric进行作业预警监控；用户可自定义Metric用于监控与预警；支持AtLeast Once 与Exactly Once语义。上文讲到平台设计要易用，下面讲平台的容错，确保数据一定不能出问题。

6.平台设计——容错

Jstorm：基于Acker机制确保At Least Once；

Spark Streaming：基于Checkpoint实现Exactly Once、基于Kafka Offset回溯实现At Least Once；

Flink：基于Flinktwo-phase commit + Kafka 0.11事务性支持实现Exactly Once。

7.Exactly Once

1）Direct Approach

当前大部分拿Spark Streaming消费Kafka的话，都是用Direct Approach的方式：

优点：记录每个批次消费的Offset，作业可通过offset回溯；

缺点：数据存储与offset存储异步：

数据保存成功，应用宕机，offset未保存 （导致数据重复）；offset保存成功，应用宕机，数据保存失败 （导致数据丢失）；2）CheckPoint

优点：默认记录每个批次的运行状态与源数据，宕机时可从cp目录恢复；

缺点：

1. 非100%保证ExactlyOnce；

https://iteblog/archives/1795 描述了无法保证Exactly once的场景；https://issues.apache.org/jira/browse/SPARK-17606 也存在doCheckPoint时出现块丢失的情况；2. 启用cp带来额外性能影响；

3. Streaming作业逻辑改变无法从cp恢复。

适用场景：比较适合有状态计算的场景；

使用方式：建议程序自己存储offset，当发生宕机时，如果spark代码逻辑没有发生改变，则根据checkpoint目录创建StreamingContext。如果发生改变，则根据实现自己存储的offset创建context并设立新的checkpoint点。

8.平台设计——监控与告警

如何能够第一时间帮用户发现作业问题，是一个重中之重。

集群监控

服务器监控：考量的指标有Memory、CPU、Disk IO、Net IO；平台监控：Ganglia；作业监控

基于实时计算框架原生Metric系统；定制Metrics反应作业状态；采集原生与定制Metrics用于监控和告警；存储：Graphite展 现：Grafana 告警：Appmon；我们现在定制的很多Metrics当中比较通用的是：

Fail：定期时间内，Jstorm数据处理失败数量、Spark task Fail数量；Ack：定期时间内，处理的数据量；Lag：定期时间内，数据产生与被消费的中间延迟（kafka 2.0基于自带bornTime）。携程开发了自己告警系统，将Metrics代入系统之后基于规则做告警。通过作业监控看板完成相关指标的监控和查看，我们会把Flink作为比较关心的Metrics指标，全都导入到Graphite数据库里面，然后基于前端Grafana做展现。通过作业监控看板，我们能够直接看到Kafka to Flink Delay（Lag），相当于数据从产生到被Flink作业消费，中间延迟是62毫秒，速度相对比较快的。其次我们监控了每次从Kafka中获取数据的速度。因为从Kafka获取数据是基于一小块一小块去获取，我们设置的是每次拉2兆的数据量。通过作业监控看板可以监控到每次从Kafka拉取数据时候的平均延迟是25毫秒，Max是 760毫秒。

接下来讲讲我们在这几年踩到的一些坑以及如何填坑的。

三、踩坑与填坑

坑1：HermesUBT数据量大，埋点信息众多，服务端与客户端均承受巨大压力；

解决方案：提供统一分流作业，基于特定规则与配置将数据分流至不同topic。

坑2：Kafka无法保证全局有序；

解决方案：如果在强制全局有序的场景下，使用单Partition；如果在部分有序的情况下，可基于某个字段作Hash，保证Partition内部有序。

坑3：Kafka无法根据时间精确回溯到某时间段的数据；

解决方案：平台提供过滤功能，过滤时间早于设定时间的数据（kafka 0.10之后每条数据都带有自己的时间戳，所以这个问题在升级kafka之后自然而然的就解决了）。

坑4：最初，携程所有的Spark Streaming、Flink作业都是跑在主机群上面的，是一个大Hadoop集群，目前是几千台规模，离线和实时是混布的，一旦一个大的离线作业上来时，会对实时作业有影响；其次是Hadoop集群经常会做一些升级改造，所以可能会重启Name Node或者Node Manager，这会导致作业有时会挂掉；

解决方案：我们采用分开部署，单独搭建实时集群，独立运行实时作业。离线归离线，实时归实时的，实时集群单独跑Spark Streaming跟Yarn的作业，离线专门跑离线的作业。

当分开部署后，会遇到新的问题，部分实时作业需要去一些离线作业做一些Join或 Feature的操作，所以也是需要访问主机群数据。这相当于有一个跨集群访问的问题。

坑5：Hadoop实时集群跨集群访问主机群；

解决方案：Hdfs-site.xml配置ns-prod、ns双重namespace，分别指向本地与主机群；

Spark配置spark.yarn.access.namenodes or hadoopFlieSystems

坑6：无论是Jstorm还是接Storm都会遇到一个CPU抢占的问题，当你上了一个大的作业，尤其是那种消耗CPU特别厉害的，可能我给它分开了一个Worker，一个CPU Core，但是它最后有可能会给我用到3个甚至4个；

解决方案：启用cgroup限制cpu使用率。

四、应用场景

1.实时报表统计

实时报表统计与展现也是Spark Streaming使用较多的一个场景，数据可以基于Process Time统计，也可以基于Event Time统计。由于本身Spark Streaming不同批次的job可以视为一个个的滚动窗口，某个独立的窗口中包含了多个时间段的数据，这使得使用SparkStreaming基于Event Time统计时存在一定的限制。一般较为常用的方式是统计每个批次中不同时间维度的累积值并导入到外部系统，如ES；然后在报表展现的时基于时间做二次聚合获得完整的累加值最终求得聚合值。下图展示了携程IBU基于Spark Streaming实现的实时看板。

2.实时数仓

1）Spark Streaming近实时存储数据

如今市面上有形形色色的工具可以从Kafka实时消费数据并进行过滤清洗最终落地到对应的存储系统，如：Camus、Flume等。相比较于此类产品，Spark Streaming的优势首先在于可以支持更为复杂的处理逻辑，其次基于Yarn系统的资源调度使得Spark Streaming的资源配置更加灵活，用户采用Spark Streaming实时把数据写到HDFS或者写到Hive里面去。

2）基于各种规则作数据质量检测

基于Spark Streaming，自定义metric功能对数据的数据量、字段数、数据格式与重复数据进行了数据质量校验与监控。

3）基于自定义metric实时预警

基于我们封装提供的Metric注册系统确定一些规则，然后每个批次基于这些规则做一个校验，返回一个结果。这个结果会基于Metric sink吐出来，吐出来基于metrics的结果做一个监控。当前我们采用Flink加载TensorFlow模型实时做预测。基本时效性是数据一旦到达两秒钟之内就能够把告警信息告出来，给用户非常好的体验。

五、未来规划

1.Flink on K8S

在携程内部有一些不同的计算框架，有实时计算的，有机器学习的，还有离线计算的，所以需要一个统一的底层框架来进行管理，因此在未来将Flink迁移到了K8S上，进行统一的资源管控。

2.Muise平台接入Flink SQL

Muise平台虽然接入了Flink，但是用户还是得手写代码，我们开发了一个实时特征平台，用户只需要写SQL，即基于Flink的SQL就可以实时采集用户所需要的模型里面或者用到的特征。之后会把实时特征平台跟实时计算平台做进行合并，用户最后只需要写SQL就可以实现所有的实时作业实现。

3.Jstorm全面启用Cgroup

当前由于部分历史原因导致现在很多作业跑在Jstorm上面，因此出现了资源分配不均衡的情况，之后会全面启用Cgroup。

4.在线模型训练

携程部分部门需要实时在线模型训练，通过用Spark训练了模型之后，然后使用Spark Streaming的模型，实时做一个拦截或者控制，应用在风控等场景。

—end—

当今互联网早已不是蛮荒生长的时代，各大公司也在自己的领域深耕细作，伴随着市场的成熟，如何提供更好的服务、更快的数据决策，成为竞争的关键点。实时计算技术作为其中的一项关键技术，开始在业界广泛流行。如何依据自己的业务，在众多的实时计算技术中做出选择，如何处理实时计算中遇到的各种问题，保证数据的效率和正确，成为所有人都要面对的极具挑战的工作。本次分享将会从实时计算的业务要求出发，结合具体的一个基于spark streaming的实践例子，总结大数据实时处理架构设计上需要处理的几个关键问题，同时基于此对实时计算技术提出一些要求。

朱健，京东广告部大数据技术专家，长期从事大数据技术的实践和研究工作，在分布式系统架构设计、开发方面有丰富的实践经验。目前负责京东广告日志系统、广告实时效果系统的研发工作。

在过去几年，对于 Apache Kafka 的使用范畴已经远不仅是分布式的消息系统：我们可以将每一次用户点击，每一个数据库更改，每一条日志的生成，都转化成实时的结构化数据流，更早的存储和分析它们，并从中获得价值。同时，越来越多的企业应用也开始从批处理数据平台向实时的流数据数据平台转移。本演讲将介绍最近 Apache Kafka 添加的一些系统架构，包括 Kafka Connect 和 Kafka Streams，并且描述一些如何使用它们的实际应用体验。

流处理

在流处理刚被提出来的时候，很多人认为流处理只能进行做近似的结果或者增量的计算，倘若你想保证其安全性，以 Lamda 架构为基础，利用流处理得到最现在的结果。但同时你需要采用 batch processing 等其他方式来保证其全局的安全性以正确性。

在如此多年的研究结果下，在我看来，流处理并不一定是近似的，或者是仅仅以无法保证真确性为代价而提高速度的一种数据处理方式。相反，流处理应该是一个与全局计算、batch processing 稍微有点不同的计算模型。跟批量处理不同之处在于，批量处理将数据引向计算，而流处理将计算引向数据。这句话大概有点模糊，接下来，我举几个大家熟悉的计算模型例子。

第一个计算模型例子—请求应答模型。

请求应答模型是业务生活中最常用的模型例子。首先提交一个请求到服务方，而服务方可能是一个数据库、也可能是别的存储工具；然后进行等待…等待；最后得到一个回答。这便是一次请求、一次计算、一次回答。该模型非常简单、也极易操作，当你需要延展到多个机器上时，只要简单地增加客户端以及处理器即可成功。但是缺点在于，不能达到大的吞吐量，每提交一次请求，都需要等待时间来获得最终应答的结果。

第二种常见的模型就是批量处理如上图所示。如果请求应答模型在谱系的一端，那么 typo 的另一端则认为是批量处理。当我积累数据数量足够多的时候，一次性提交任务到数据仓库，再进行等待，等待时间短则几秒钟、几分钟，长则几小时，最后才得到最终的结果—所有输入对应的所有输出。该批处理模型的好处在于能够提高其吞吐率，一次的请求和应答可以得出较多结果。但它的缺点是具有高延时性，比如某数据产生时间为上午 6 点钟，用户点击某网页，由于批处理模型，每 12 小时才会运行一次，那么它必须等到上午 6 点到下午 6 点的所有数据完整以后才会进行工作，那么运行结果可能是用户点击的 12 个小时之后。高延迟性是批处理自身带有的特性。

那么什么是流处理呢? 在我看来，流处理就是介于请求应答和批处理之间的一种新型计算模型或者编程模型。流处理并不等待数据的完整性，或者说数据本没有完整性这一讲法，数据本身就是一个数据流，当每个数据流每产生一个新数据的时候立刻被计算出、进行返回，因此数据是源源不断地通向计算，并且源源不断有结果被输出。你可以设想，与等待数据完全完成之后发布到计算上相比，流处理就是将计算移到你数据发生地进行实时计算的方式。

为什么很多人之前有这样一种错觉，他们认为流处理可能存在有丢包的情况、或者说只可以得到近似的结果，其实这是早期的一些数据流处理系统所自带的一些限制。因此以 Lamda 架构为基础，在流处理上需要讨论不同维度的取舍。接下里我将举三个例子，延迟、、成本和正确性。正如很多人之前提及的，在进行流处理时候，其大多数情况需要用时间来换取正确性，或者用更多的成本换取时间等等。

第一个例子，说如果你需要做一个实时的 ETL 处理。而关于 ETL 处理不需要太小的延迟，为达到低成本的一种保证，我们可以忍受几分钟或者 1 分钟的延迟；但是，如果你正在进行一个实时的在线监测，存在着几毫秒的延迟，那么这时候可能更愿意选择花大量的金钱，或者采取一些可能不必要的 possibility 来达到一种低延迟的效果；第二个例子，假设你在做一个在线付费协议，它也是一个流处理平台。由于在线付费协议可能关乎到其机构，或者其公司的利益所在，因此你会说，我需要保证百分之百的正确性，我不希望有任何丢包情况；

第三个例子，如果你是做一个实时的日志处理，实时收集所有日志，并将其导入 root，在这种情况下，你可能会说，为了降低成本，我愿意付出一小部分正确性的代价，即使不能达到 100%、达到 99.99%、达到 99.9%，这样的结果都可以接受。这本是用户在定义不同流处理应用或者业务的时候应该可以自己做出的选择。但比较遗憾的是，多数早期的流处理平台其实并没有给予用户该种选择，他们自身的设计理念，那就是为了低延迟直接放弃掉正确性，或者说为了更高的吞吐量直接放弃低延迟。

以上是我想分享的关于流处理的一些误会认知，如果我的分享能够让大家带走两个答案的话，我希望这就是一个。我认为流处理仅仅是一种不一样的计算模型或者编程模型，它将计算带到数据上，而不是将数据引用到计算上，并且在流处理的时候，用户往往需要在正确性、延迟性、成本等不同的维度上做出选择。

Kafka 的角色

为什么当我们说到流处理的时候，很多人都在说 Kafka。大多数人在最早接触 Kafka 时会说，Kafka 就是一个分布式发布订阅的消息系统，但是如果我们去观察 Kafka 的最初一些设计特性可发现以下几点内容。第一点，它可以作为一个写在磁盘上的缓存来使用，或者说，并不是仅基于内存来存储流数据，它可以保证数据包不被及时消费时，依然可用且不被丢失；第二点，由于位移的存在提供了逻辑上的顺序，在同一个话题上，第一个数据比第二个数据最先被发布的时候，也可保证在消费时也是永远第一个数据比第二个数据先被消费；第三点，因为 Kafka 是一个公有的大数据中转站，就是说，所有的数据只要在 Kafka 上，永远可以在 Kafka 周围进行业务的开发或者认知事物的开发。接下来我将花费一些时间详细介绍这三点之间的关系。

Kafka 不仅仅是一个订阅消息系统，同时也是一个大规模的流数据平台，那么它提供了什么呢？第一，提供订阅和发布消息；第二，提供一个缓存的流数据存储平台；第三，提供流数据的处理平台。今天，我将着重讨论流式计算在 Kafka 上面的应用。

流式计算在 Kafka 上的应用主要有哪些选项呢？第一个选项就是 DIY，Kafka 提供了两个客户端 —— 一个简单的发布者和一个简单的消费者，我们可以使用这两个客户端进行简单的流处理操作。举个简单的例子，利用消息消费者来实时消费数据，每当得到新的消费数据时，可做一些计算的结果，再通过数据发布者发布到 Kafka 上，或者将它存储到第三方存储系统中。DIY 的流处理需要成本。打个比方，考虑数据的延迟性，考虑不同时间上的管理分配，正如很多人提到的 processing time，这将是我后文会重点提及的概念。以上这些都说明，利用 DIY 做流处理任务、或者做流处理业务的应用都不是非常简单的一件事情。

第二个选项是进行开源、闭源的流处理平台。比如，spark。关于流处理平台的一个公有认知的表示是，如果你想进行流处理操作，首先拿出一个集群，且该集群包含所有必需内容，比如，如果你要用 spark，那么必须用 spark 的 runtime。因为他们划定了你作为一个流处理平台使用者需要用到的所有行为，比如，资源管理系统、参数调配系统、容器配置、代码封装、分发等，以上行为都已被该平台所限定。一旦你选择使用甲就必须用甲套餐装备，如果选择使用乙就必须使用乙套餐装备。有人不禁提出疑问，我能不能既选择流处理平台，又使用自己选择的，我能不能这样做呢？

这个应用场景其实很普遍，举个例子，可异步式微服务处理。什么叫异步式微服务处理？假设 Kafka 作为一个缓存数据，在该缓存区含有很多不同的业务。打个比方，一个网店的机构可以有不同的组、不同的员工，有人负责销售、有人负责商品分发，有人负责价格管理、有人负责在线实时的限流监控，不同的组、不同的员工可能会以不同的时间，或者以不同的代码来更新他们的产品，只要拥有一个异步式缓存机制，即 Kafka，便可扩大该微服务，而不需要他们的任何一个组之间进行同步请求应答机制。

在该微服务情况下，每个小组的喜好、特性并不一致，有的组表示我需要做流处理平台，从 Kafka 读数据，处理完再写回 Kafka，并且想要使用 EWS 把我的应用部署在云端大规模集群上；而另外小组表示我不需要那么复杂，我只是小规模数据，不希望起一个集群，只需起三个机器，并且每个机器有 1GB 内存足以，可进行手动控制操作，不需要资源管理器。那么我们能不能同时满足他们不同的需求呢? 答案就是我接下来要说的第三种选项。

第三种选项是使用一个轻量级流处理的库，而不需要使用一个广泛、复杂的框架或者平台来满足他们不同的需求。在 Kafka 0.10 当中已发布轻量级流处理内容平台，我们可以设想，跟其他客户端发布者和消费者一样，它也是一个客户端，不同之处在于它是一个计算者客户端，一个好用的、功能强大的客户端，并且支持 state processing、Windows 延时的、异步的、甚至不同数据的调控。 最重要的是 Kafka 作为一个库，可以采用多种方法来发布流处理平台的使用。比如，你可以构建一个集群；你可以把它作为一个手提电脑来使用；甚至还可以在黑莓上运行 Kafka。以上都是尤其简单的运行库的概念。

因此我们要做的事情与使用 Kafka 其他的客户端类似，比如发布者、消费者，只要在代码里边加入就可以使用各种各样的 API。当你要调配控制 Kafka Stream 应用的时候，选择最基础的 War File 来运行或者采用 Java、C，甚至资源管理器来运行都是可行的。因为 Kafka Stream 是一个轻量级流处理的库，可支持各种各样的运维方式。

在我们看来，简单的就是美的，只有给用户提供最大的兼容性与最大的延展性，用户才能得到最好的用户体验。

Kafka Stream 的编程语言

如果接触过 Storm、Spark 等流处理平台的同学可以发现，它们与 Kafka Stream 高阶位 DSL 语言其实有相似之处。如上图所示，首先定义一个 Streams 流， Streams 是从 topic1 中的 topic 获取得到，即定义 Streams、处理 Streams、得到新的 Streams。比如，从 topic1 里面得到两个原始数据流，然后数据流进行 countByKey 得到新的数据流叫做 Counts。那么 counts.to(“topic2”) 是什么意思呢？在获取到新的数据流之后写回 Kafka topic2 内，启动 KafkaStreams 进程，与 Kafka producer、Kafka consumer 类似，让它来运行已定义计算。

正如大家所了解的，API 的使用其实很简单。提供一个简单的 API，用户简单地写入运行逻辑即可运行。但是编程应用总是容易的，而它的复杂程度在于，一旦你开始运维该应用，当你想要把业务拓展到更大规模，或者业务出现变化，或者集群不稳定，需要强大的运维时，运维的程度便显得异常重要，最上面的编程可能只是冰山一角。Kafka Stream 的设计理念是最简单的就是最美的，包括 API、运维、debugging，以及各种各样的方式，都是希望给用户带来最简单的体验。它的核心思想就是把难问题直接给 Kafka 集群本身。

Kafka 的介绍

Kafka 的核心思想是什么？就是把这些消息全部存成一个有序日志，所有的消息发布者把消息发布到底端，从某一个逻辑上的位移开始顺序读取所有的消息。它的一个好处在于所有的读和写，尽管都是刷到磁盘上，但都是按照顺序进行，该方式对磁盘的使用比较有效，倘若消费者和发布者隔得比较近，将利用 page cash 直接读数据。

延展性。如上图，提供 topic 以及 topic partitions，即话题与话题分区的机制。每个用户有不同的 topic，每个 topic 可以有多个分区，每个分区可被装载在不同的机器上，当用户提高规模之后，Kafka 只需要简单地增加机器和 topic partitions 数量，或者采用 ROM balance 的方式到不同机器上，即可达到线性延展方式。

以上是 Kafka 最简单的核心思想，接下来我将介绍 Kafka Streams 作为 Kafka 客户端如何利用以上核心思想来设计流处理的平台。数据流其实就是有序的记录或消息，每个消息是一个 Key 加一个 Value，并且 record 与 Kafka 自身 massage 具有一一对应关系。

用户所提供的业务上的计算模型，其实可用拓补结构进行表达。如上图，图的左边。用户首先进行定义数据流，然后对数据流进行计算，得到新的数据流，最终将数据流写回到 Kafka 内。每当用户进行定义的时候，每一步都会变成拓扑结构里面的一个点，每个点通过流进行计算，变成新的流来进行新的连接，最终在 Kafka 内部形成拓扑结构。用户并不需要在意该拓补结构，只需明白定义流、计算流、得到新的流，写回 Kafka。

连接每一个不同的运算单元就是一个 Stream，即 record stream，每一个 Stream 都在源源不断地实时产生 record，每一个 record 是一个 key 加一个 value。利用 Stream Processor 连接 Stream，每个用户定义的流的一个计算单位对应着一个 Stream Processor。

当用户定义每一步计算的时候，就是定义每个拓扑结构里面的每个点，最终把整个拓补结构定义完整到 Kafka Stream 来运行。计算单�...

1 引言

大数据技术的广泛应用使其成为引领众多行业技术进步、促进效益增长的关键支撑技术。根据数据处理的时效性，大数据处理系统可分为批式(batch)大数据和流式(streaming)大数据两类。其中，批式大数据又被称为历史大数据，流式大数据又被称为实时大数据。

目前主流的大数据处理技术体系主要包括Hadoop[1]及其衍生系统。Hadoop技术体系实现并优化了MapReduce[2]框架。Hadoop技术体系主要由谷歌、推特、脸书等公司支持。自2006年首次发布以来， Hadoop技术体系已经从传统的“三驾马车”(HDFS[1]、MapReduce和HBase[3])发展成为包括60多个相关组件的庞大生态系统。在这一生态系统中，发展出了Tez、Spark Streaming[4]等用于处理流式数据的组件。其中，Spark Streaming是构建在Spark基础之上的流式大数据处理框架。与Tez相比，其具有吞吐量高、容错能力强等特点，同时支持多种数据输入源和输出格式。除了Spark开源流处理框架，目前应用较为广泛的流式大数据处理系统还有Storm[5]、Flink[6]等。这些开源的流处理框架已经被应用于部分时效性要求较高的领域，然而在面对各行各业实际而又差异化的需求时，这些开源技术存在着各自的瓶颈。

在互联网/移动互联网、物联网等应用场景中，个性化服务、用户体验提升、智能分析、事中决策等复杂的业务需求对大数据处理技术提出了更高的要求。为了满足这些需求，大数据处理系统必须在毫秒级甚至微秒级的时间内返回处理结果。以国内最大的银行卡收单机构银联商务为例，其日交易量近亿笔，需对旗下540多万个商户进行实时风险监控，在确保这些商户合规开展收单业务的同时，最大限度地保障个人用户的合法权益。这样的高并发、大数据、高实时应用需求给大数据处理系统提出了严峻的挑战。银联商务以前使用的T+1事后风控系统存在风险侦测迟滞高(次日才能发现风险，损害已经造成)、处理时间长(十几个小时之后才能完成风险识别)、无法处理长周期历史数据(只能分析最近几日的流水数据)以及无法支持复杂规则(仅能支持累积求和等简单规则)等重大缺陷。为此，亟须研发全新的事中风控系统，以重点实现低迟滞(在1 min内甄别突发风险)、高实时(100 ms内返回处理结果)、长周期(可处理长达10年以上的历史周期数据)以及支持高复杂度规则(如方差、标准差、K阶中心矩、最大连续统计等)等目标。这一目标可以抽象为一个大数据处理科学问题：如何在一个完整的大数据集上，实现低迟滞、高实时的即席(Ad-Hoc)查询分析处理。

2 技术解析

现有的大数据处理系统可以分为两类：批处理大数据系统与流处理大数据系统。以Hadoop为代表的批处理大数据系统需先将数据汇聚成批，经批量预处理后加载至分析型数据仓库中，以进行高性能实时查询。这类系统虽然可对完整大数据集实现高效的即席查询，但无法查询到最新的实时数据，存在数据迟滞高等问题。相较于批处理大数据系统，以Spark Streaming、Storm、Flink为代表的流处理大数据系统将实时数据通过流处理，逐条加载至高性能内存数据库中进行查询。此类系统可以对最新实时数据实现高效预设分析处理模型的查询，数据迟滞低。然而受限于内存容量，系统需丢弃原始历史数据，无法在完整大数据集上支持Ad-Hoc查询分析处理。因此，研发具有快速、高效、智能且自主可控特点的流式大数据实时处理技术与平台是当务之急。

实现一个融合批处理和流处理两类系统且对应用透明的系统级方案，需要攻克以下几个技术难点。

(1)复杂指标的增量计算

尽管计数、求和、平均等指标能够依靠查询结果合并实现，然而方差、标准差、熵等大部分复杂指标无法依靠简单合并完成查询结果的融合。再者，当查询涉及热点数据维度及长周期时间窗口的复杂指标时，多次重新计算会带来巨大的计算开销。

(2)基于分布式内存的并行计算

采用粗放的调度策略(例如约定在每天的固定时间将流数据导入批处理系统)会造成内存资源的极大浪费，亟须研究实现一种细粒度的基于进度实时感知的融合存储策略，以极大地优化和提升融合系统的内存使用效率。

(3)多尺度时间窗口漂移的动态数据处理

来自业务系统的数据查询请求会涉及多种尺度的时间窗口，如“最近5笔刷卡交易的金额”“最近10 min内密码重试次数”“过去10年的月均交易额”等。每次查询请求都重新计算结果会对系统性能造成极大的影响，亟须研究实现一种支持多种时间窗口尺度(数秒到数十年)、多种窗口漂移方式(数据驱动、系统时钟驱动)的动态数据实时处理方法，以快速响应来自业务系统的即席查询请求。

(4)高可用、高可扩展的内存计算

基于内存介质能够大大提升数据分析及处理能力，然而由于其易挥发的特性，一般需要采用多副本的方式来实现基于内存的高可用方案，这使得“如何确保不同副本的一致性”成为一个待解决的问题。此外，在集群内存不足或者部分节点失效时，“如何让集群在不间断提供服务的同时重新平衡”同样是一个待解决的技术难题。亟须研究分布式多副本一致性协议以及自平衡的智能分区算法，以进一步提升流处理集群的可用性以及可扩展性。

“流立方”流式大数据实时处理技术在上述领域取得了一系列突破，该技术提供基于时间窗口漂移的动态数据快速处理，支持计数、求和、平均、最大、最小、方差、标准差、K阶中心矩、递增/递减、最大连续递增/递减、唯一性判别、采集、过滤等多种分布式统计计算模型，并且实现了复杂事件、上下文处理等实时分析处理模型集的高效管理技术。

3 平台纵览

基于“流立方”流式大数据实时处理技术，研发了“流立方”流式大数据实时处理平台。其应用框架如图1所示，具有良好的灵活性和适应性。平台的数据装载模块负责从具体业务系统中接入实时流数据，数据抽取模块负责批量抽取历史数据，模型装载模块负责将分析处理模型集中的计算模型和脚本加载到平台中。当收到业务系统发出的实时查询请求时，“流立方”平台能够根据分析处理模型在完整大数据集上实时计算出相应的指标，并进行判断，将结果反馈给业务系统。

图1 “流立方”平台应用框架

在测试环境为8台服务器(每台服务器配置24核 CPU、256 GB内存)，同时计算16个统计指标(涉及4个维度，包含计数、求和、平衡、最大、最小、标准差、过滤、去重、排序、复杂事件处理等多种算法)的性能测试中，“流立方”平台达到了单节点写入大于43 000 TPS、8节点读取大于100万TPS、平均时延为1～2 ms的优异性能，如图2所示。

图2 “流立方”平台性能指标

“流立方”平台在解决批式大数据和流式大数据融合实时处理技术难题，实现优异性能的同时，还解决了流式大数据处理平台面临的两大工程化难题。一是作业的编排效率问题。大部分开源流处理平台在完成一个流处理编排时，都需要经过拓扑设计、代码编写、功能测试、打包部署等环节，一般需要一周的时间才能完成。“流立方”平台通过基于“所见即所得”的在线作业编排管理，将上线任务耗时降低到分钟级，大大提升了流处理作业的编排效率。二是流处理作业的灵活变更问题。流处理平台擅长进行逻辑预先定义的增量计算，尽管其计算效率极高，但计算灵活度受到限制。例如，某业务需要统计过去3个月的数据，现有的流处理平台在该业务上线3个月后才能完全生效，这样的工作方式使流处理技术在实际应用中受到很大的局限。“流立方”平台创新性地引入流媒体播放器的录制与重放思路，在原始数据进入流处理平台时，通过顺序写的方式持久化一份原始数据，在需要上线新的计算作业时，即刻重发指定时间窗口内的原始数据，从而实现快速(分钟级甚至秒级)计算作业上线。

“流立方”平台引入了一系列创新技术，在性能、可用性、可扩展性等多个层面提升了流处理平台的处理能力，满足金融领域在内的众多领域的业务及运维需求。引入数据冲突智能规避技术，解决了流式处理中的热点数据处理问题，从而解决了大颗粒数据维度的处理效率问题;引入Paxos一致性协议，解决内存存储计算时多副本一致性问题，提供了面向运维人员透明的一致性解决方案;引入智能分区技术，基于一致性散列技术，进一步将散列值拆解为散列块，通过散列块的平滑迁移解决存储集群的可伸缩性设计问题，确保对于运维人员的集群变更透明性;引入计算作业的动态运行时加载技术，规避了作业手工打包部署的问题，进一步提升了开发人员的工作效率。

在国内某大型银行卡收单机构组织的招标测试中，测试环节为两台低配置虚拟机，测试数据为该机构的数千万笔交易流水，计算逻辑包括50多条规则，涉及30多个统计指标。在该测试环节下，两家国外著名厂商中，一家厂商的计算时间长达24 h，另一家老牌数据库软件提供商则未能在一天内完成计算。相较于这些国外著名厂商的大数据处理平台，“流立方”平台能够在3 h内完成所有计算，且正确率为100%。

4 应用场景

“流立方”流式大数据实时处理系统在金融、交通、电信、公安等行业具有广泛的应用场景。以金融风控反欺诈为例，部署“流立方”风控系统仅需在交易前端增加风控探头，将实时交易数据旁路接入系统。“流立方”风控系统根据融合了专家知识和机器学习结果的数百条规则对每笔交易进行风险评估，判断是否允许进行该笔交易，流程如图3所示。该系统平均响应时间在6 ms以下，并发数超过50 000笔/s。同时，实现这一性能仅需要4台服务器。

图3 基于“流立方”的金融风控反欺诈流程

基于“流立方”的金融风控反欺诈技术体系包含技术(如设备指纹、代理侦测、生物识别、关联分析、机器学习等技术)、知识(如盗卡反欺诈、伪卡反欺诈、信用卡套现、营销反欺诈等规则与模型)、数据(如虚假手机数据、代理IP数据、P2P失信数据等标识数据)三大板块。技术部分中的设备指纹技术通过主被动混合的形式采集设备中软硬相关要素，结合概率论等算法为每一个设备颁发一个全球唯一的指纹编码，这些指纹编码在反欺诈的整个过程中起到非常积极的作用;代理侦测技术通过短时间内扫描IP相关端口来识别那些开启代理的IP，并在这些IP访问金融服务时进行识别;生物识别技术通过采集设备上用户的鼠标点击、触摸、键盘敲击等行为识别操作者是人还是机器以及是否操作者本人的问题;关联分析技术在底层通过图数据库存储不同节点以及关系信息，最终在界面上通过图的形式进行欺诈者关联分析及复杂网络分析;机器学习技术通过有监督、无监督的机器学习算法提升欺诈识别的准确率及覆盖率，并结合流立方技术提供模型的事中预测能力。

基于上述技术体系，研发了银行业务风险实时监控系统、互联网支付业务风险实时监控系统、电商业务风险实时监控系统等金融风控反欺诈系列解决方案。这些方案已应用到银行、第三方支付机构、互联网金融等领域的上百家企业。目前50%以上的线下交易都在“流立方”的保护下进行，基于“流立方”的金融风控反欺诈解决方案每天为我国的金融机构抵御上亿次的攻击。该技术已经成为我国金融安全领域基础设施必不可少的组成部分。

此外，在互联网机器防御系统中，“流立方”同样能发挥巨大作用。如今网络机器人遍布票务、电商、招聘、银行、政府、社交等各类网站，消耗了40%～60%的网络流量。网络机器人不仅消耗网络资源、影响正常客户访问、增加网站运营成本，还会爬取产品、价格信息，形成不正当竞争，甚至混淆网站用户生态，影响营销分析。传统的控制策略通过采取屏蔽频繁访问、设置验证码等方式防御网络机器人，无法应对日益智能化的新型网络机器人。基于“流立方”的互联网机器防御系统通过在Web服务器上嵌入插件或者独立的嗅探器(sniffer)程序，将全流量的Web访问请求旁路到独立的机器防御集群，进行实时的流量分析及防御决策，并将决策后的结果实时回馈到Web服务器插件中。Web服务器插件在判定当前访问的设备或者IP地址等是机器人时，能够自动改写响应内容，根据不同的风险级别自动拒绝交易或将访问者引导到第三方图形验证码服务商进行机器人验证。访问者在通过验证后可以继续正常访问Web服务。该系统还创新地将设备指纹以及人机识别服务运用到机器防御系统中，不仅增加了可分析维度，提升了控制颗粒度，同时能够对基于浏览器内核的高级爬虫进行防护。此外，将机器防御规则、数据服务、设备指纹、人机识别以及图形验证码以软件即服务(software as a service，SaaS)的形式提供服务，进一步降低了互联网网站客户的运维门槛，提升了产品竞争力。该机器防御系统工作过程如图4所示。

基于“流立方”的实时机器防御系统通过多服务器访问流水关联决策、长周期数据决策、复杂规则爬虫识别、设备维度爬虫识别、人机识别等技术，实现了微秒级(400～800μs)的识别时延，同时具有机器人识别管控一体化、轻量级接入等优点。根据已经接入机器防御服务的几十家客户的反馈，基于“流立方�...

数据时代，大数据计算已经渗透到了各行各业，业务沉淀数据，数据计算产生新的业务价值，大数据计算正不断地用这种方式推动业务向前发展。电商双11，商家与消费者狂欢的背后，同样离不开大数据计算带来的价值贡献，特别是应用越来越广泛的“实时计算”。

现实世界中，数据连续产生，并被实时采集和计算

我们要做数据计算，挖掘产品商业价值，首要解决的问题是数据的问题。现实世界里，数据往往是随着时间的推进连续产生的，比如用户浏览商品，一系列的鼠标点击操作，会产生一连串的后台数据；开车使用手机导航，GPS定位每隔一段时间更新一次，也会不断产生日志数据；用户浏览新闻推送、搜索歌曲、监控摄像头定时采集图片上传到云端存储、视频直播等等场景，这背后生成的数据都是连续产生的。连续产生的业务数据，又被实时采集起来，就形成了数据流。

流式数据一经采集，就可以立即参与计算，同时将计算结果投入到业务应用中，这就是实时计算。实时数据计算其实早已经进入到人们生活的方方面面了，比如天气预报，以前人们的习惯是每天接收一次天气预报信息，现在则可以实时查看天气预测，同一个时间点的天气预测会随着时间的接近越来越准确，这就是监测数据采集更新及实时数据计算带来的效果。

根据兴趣量身定制，实时计算让产品越来越了解用户

实时数据来源越来越多、数量越来越大，每年的数据量都在成倍地增长，这对实时计算本身是利好的，可以有更多的应用场景、更好的应用效果，还可能促成一些革命性的变化。那么，大数据实时计算还能做什么？

在网易，考拉海购双11、618海淘盛典等活动期间，都会有一块网易有数大屏幕实时展示当前最新的销售总额、每个商品品类的销售比例、订单增长趋势、活跃用户地理位置等，各种维度的信息都在一块屏幕上不断跳动。每个用户每笔订单所产生的影响都会实时更新到大屏上。这种可视化的实时应用效果，除了增添一份电商狂欢节的氛围，更易于发现数据价值，指导市场运营、辅助商业决策。

金融风控是另一种典型的实时计算应用场景。对金融业务这种风险敏感的业务来说，仅仅能把数据可视化是远远不够的，它需要流计算系统能够利用一些风险模型的匹配规则，去实时分析海量的用户行为数据，发现异常事件、判断风险等级，并作出相应的风险控制措施，自动化地去做报警通知、改变业务流程。通过实时计算做金融风控，带来的好处是更快、更准、更广。其他许多类似风控这样的事件驱动计算场景，实时计算都能解决好。

实时计算在推荐领域的应用也已经很深入了。不论是新闻推荐、音乐推荐还是读书推荐，基本都已经做到了千人千面，每个人接收到的推送内容都是根据个人兴趣偏好量身定制的。而用户的兴趣偏好，往往是通过实时数据计算不断在更新的。 以新闻推送为例，当用户点击一条条推送消息时，背后产品其实时刻在对用户的行为做实时分析，实时更新用户的兴趣偏好，不断发现用户新的兴趣点，对用户越来越了解，最后给用户推送他更感兴趣的内容。再以音乐推荐为例，如果一个用户某段时间收藏了几首悲伤的歌曲，通过实时数据分析，系统可以识别出这一信息，同时有针对性的推送一些歌曲去抚慰用户。这种场景是只有实时计算才能解决的，也最能体现实时计算的价值。

越来越多的实时计算场景会被开发出来，未来人们对“一切都在变化之中”的感受会越来越深刻。

从“先存后算”到“边算边存”，实时计算不再怕“大”数据

实时计算这么好，在实现层面应该怎么做，有哪些困难和挑战是必须解决的？

首先从整体架构看，数据计算，无外乎三样东西：数据输入→计算→数据输出。传统的计算模型，以数据库为例，是先将数据存储在一个数据表中，用户通过执行查询语句触发数据库的计算操作，最后数据库完成计算后输出结果。这种“先存后算”的模型在大数据实时计算场景下是行不通的。我们所要计算的数据很“大”，一个计算结果所涉及的源数据可能是涵盖过往一天的数据，可能是上千亿条数据记录。如果每增加一些新数据，都把所有数据都重新计算一遍，这样的开销是非常大的，最终的效果会是很“慢”，达不到实时的效果。比较合理的做法是“边算边存”，意思是数据进入实时计算系统后，不一定需要先存储起来，可以直接参与计算，而且这里的计算是把当前新增的数据在之前历史数据的计算结果上做“增量计算”，同一条数据不重复参与计算，计算完成之后，再把计算结果保存起来，供业务使用，这时数据存储的压力也小了很多。同时“大”意味着数据并发很高，每秒可能需要计算上千万条新数据，这样的计算量不是单机能承受的，所以大数据实时计算要解决好的是分布式系统架构下的一系列技术问题。

分布式实时计算面临的挑战包括很多方面。数据从采集、到计算、到输出整个过程必须做到低延迟，除了计算节点本身采用“增量计算”的模型，还要求上游数据传输模块具有很高的吞吐能力，并且具备数据缓存的能力，在大流量场景下可以起到缓冲的作用，下游输出模块也需要做数据压缩、批量输出等优化，以保证输出结果的实时性。低延迟这个大前提对实时计算系统的其他特性提出了更高的要求。比如双11凌晨0点的时候，大量消费者在同一时刻下单支付，这是涌进实时计算系统的瞬时数据量是巨大的，系统需要有强大的并行处理数据的能力，将大量瞬时流量合理分配到成百上千个计算节点，并将这些节点的计算结果汇聚到一起计算出一个总体的结果，在高吞吐的情况下仍保证低延迟。

从“批量计算”到“增量计算”，最具挑战的是准确性和易用性

和低延迟同样关键的挑战是准确性。“增量计算”模型和传统“批量计算”模型是有区别的，所以不能照搬过往的技术经验，否则就会有准确性方面的问题。需要考虑清楚新进入的数据如何叠加到老的计算结果上，有些场景下甚至要支持从老的计算结果中撤除部分计算值，以保证最终结果的准确性。

分布式系统中的某个节点出现故障是很常见的，实时流计算系统的故障恢复能力也相当重要，因为当故障发生时，系统必须快速恢复，否则系统的输出更新可能就停滞了，实时性也就无从谈起。同时故障发生也不能破坏“增量计算”这个模型，否则退化到“批量计算”的模型就又得不到实时的计算结果了，而且结果准确性也难以保证。

事实上网易大数据在实现自研流计算平台Sloth的过程中，遇到并克服了上述技术难点。网易流计算平台Sloth作为一个平台化的产品，在产品易用性、多租户隔离方面做了大量的工作。就实时计算而言，易用性是一个比较值得讨论的方面。

对于开发人员而言，写一个分布式程序比写单机程序会困难一些，而写一个分布式实时计算程序，会更难。好在业界有一些开源的流计算引擎帮助完成了不少工作，开发人员可以使用这些流计算引擎完成流计算任务的开发，他们可能不再需要关心计算任务如何分发到多个计算节点上、数据在计算节点间如何传输等问题，只需要专注于计算逻辑的开发、控制好不同计算阶段的计算并行度。

以计算一篇文章的单词数为例，一个分布式计算程序的内容可能包括三个部分，首先是用几个计算节点共同把每一行文本拆分成一个一个的单词；第二步是用另外一些计算节点去统计单词的个数（考虑到数据量巨大的情况，这里有必要用多个节点去做计算）；第三步是由一个计算节点把上游各各节点算出的部分计数汇聚成一个总的计数。这样一个最简单的场景，需要开发的代码量大约是200行。实际业务场景下，数据流经的计算节点远远不止3个，计算类型也比基础的求和复杂很多，所以即使有了流计算引擎，分布式实时计算程序的开发仍然是比较困难的。再进一步看，即使开发完成了，还需要把大量的时间投入到调试、计算框架维护等方面，一旦计算需求发生变化，所有的工作都需要重新迭代一遍，这是个比较痛苦的过程。如何让流式计算程序更易编写，是实时计算平台需要去完成的挑战。

且不考虑实时流计算系统如何解决易用性这个问题，看下计算机科学发展过程中，类似问题是怎么解决的。人们希望编程可以容易一些，所以越来越多的高级编程语言被发明出来了；人们希望数据计算可以容易一些，然后就有了数据库，以及SQL语言——结构化查询语言；到了大数据时代，人们还在折腾离线批量计算的时候，就遇到的依靠计算引擎编程复杂的问题，最终通过把SQL语言应用到分布式离线计算系统上，解决了这个问题。而现在实时计算的迅速发展的现在，是否同样可以用SQL语言去解决这个问题？答案是肯定的。不过有许多细节的问题需要去推敲求证。

实时流计算中的数据流，可以理解为一张动态的数据表

上文提及了离线批量计算模型和实时增量计算模型是有差异的，当SQL语言分别作用与批量计算和流式计算时，其语义也是需要发生变化的。批量计算和流式计算最主要的区别是前者计算的数据是有限的、后者计算的数据是无限的是不断采集进入系统的。当一个SQL查询作用在一批离线数据上面时，计算完成、输出结果，这条SQL查询也就完成了。映射到流式计算，当SQL查询触发计算，它是不会结束的，因为数据在持续不断地流入，按照离线SQL的语义，SQL结束之前，计算不会输出结果，这显然不是流计算期望的效果，所以流式SQL其本质应当是定义一系列流计算任务，同时这些任务是边执行边输出计算结果的。

离线SQL处理的是静态数据表，而流式SQL处理的是数据流，SQL的计算语义（如求和、平均值、数据表连接等）作用在数据流上是否合理。理解这个问题需要做一个概念上的转换：离线SQL是把静态的数据表转换成另一张静态数据表；而实时流计算中的数据流，可以理解为一张动态的数据表（数据会不断增长的动态数据表）。不同的时刻这个数据表又不同的样子，执行SQL会得到不同的计算结果，把这些不同的计算结果像电影幻灯片放映一样串联起来，我们就得到了一张动态的结果表——流式SQL做的工作就是把一张动态数据表转换成另一张动态数据表，这样流SQL的计算语义就比较容易理解了。实时流计算系统要解决的问题就缩小到了“如何实现动态数据表的计算”上来。

流SQL引擎的自动优化是当前主要的技术突破方向

实时流计算系统的易用性，是可以用SQL语言来解决的，网易流计算平台Sloth的生产实践也证实了这一理论。用户不再需要学习各种计算引擎的编程接口，不再需要调试分布式计算程序，不再需要自己维护流计算系统，只需要把原来跑在离线平台上的SQL迁移到实时流计算平台上，就可以完成复杂的实时计算逻辑。

用户端的工作大大减少了，实时流计算平台的工作势必是要增加的，其中比较困难的部分是如何把SQL查询转化成实际的计算逻辑，实现一个支持流式SQL的计算引擎，类似数据库引擎的角色，而且就像之前讨论的，这个引擎的计算逻辑必须符合“增量计算”模型。同时为了能让实时计算结果应用到各种各样的业务场景中，计算引擎需要能够对接各种存储角色，比如数据、消息队列、离线存储等。

双11大屏只是大数据实时流计算的一种应用场景，未来会有越来越多的实时计算场景，比如除了文本计算实时化，图像、语音计算也可以实时化，在线机器学习，物联网实时计算等。实时数据以及实时流计算场景的类型都是指数增长的，实时计算引擎会面临不小的挑战。基于SQL的流式计算描述也正在向前演化，会越来越多的纳入流计算特有的属性，比如输出触发、过期数据处理、多种规则的数据窗口划分等。流SQL引擎的自动优化也是当前主要的一个技术突破方向，相信未来实时流计算会随着技术的进步，应用得跟深入、更广泛。

[来自IT168]　　【IT168 专稿】本文根据伍翀老师在2018年5月12日【第九届中国数据库技术大会】现场演讲内容整理而成。　　讲师简介：

伍翀，阿里巴巴高级研发工程师，花名“云邪”，阿里巴巴计算平台事业部高级研发工程师，Apache Flink Committer。北京理工大学硕士毕业，2015 年加入阿里巴巴，参与阿里巴巴实时计算引擎 JStorm 的开发与设计。从 2016 年开始从事阿里新一代实时计算引擎 Blink SQL 的开发与优化。现在专注的方向主要是分布式处理和实时计算，热爱开源，热爱分享。　　摘要：　　SQL是数据处理中使用最广泛的语言。它允许用户简明扼要地声明他们的业务逻辑。大数据批计算使用SQL很常见，但是支持SQL的实时计算并不多。Apache Flink是一款同时支持批和流计算的引擎，Flink SQL的实现完全遵循ANSI SQL标准，这是它和其他流处理框架(例如Kafka和Spark)在DSL上的一个重要的不同。阿里巴巴是Flink SQL最大的贡献者，Flink开源社区的一半以上的SQL功能都是阿里工程师开发的。阿里内部绝大部分的流计算业务也都在使用Flink SQL编写。本次演讲我们将介绍Flink SQL的设计原理以及分享在阿里大规模使用中收获的经验。　　分享大纲：　　1、Background　　2、Flink SQL基本概念　　3、Flink SQL核心功能　　4、Flink SQL优化　　5、阿里云流计算产品　　正文：　　阿里巴巴自2015年开始调研开源流计算引擎，最终决定基于Flink打造新一代计算引擎，针对Flink存在的不足进行优化和改进，并将最终代码贡献给开源社区。目前为止，我们已经向社区贡献了数百个Commiter。阿里巴巴将该项目命名为Blink，主要由Blink Runtime与Flink SQL组成。Blink Runtime是阿里巴巴内部高度定制化的计算内核，Flink SQL则是面向用户的API层，我们完善了部分功能，比如Agg、Join、Windows处理等。今年，我们已经全部跑通TPCH 及TPC-DS的Query，熟悉数据库的人都知道，这代表着整个数据库或引擎是一个基本功能完备的产品。　　接下来主要介绍Flink SQL的基本概念及使用。传统的流式计算引擎，比如Storm、Spark Streaming都会提供一些function或者datastream API，用户通过Java或Scala写业务逻辑，这种方式虽然灵活，但有一些不足，比如具备一定门槛且调优较难，随着版本的不断更新，API也出现了很多不兼容的地方。　　我们一直在思考最适合流计算处理的API，毫无疑问，SQL已经成为大数据领域通用且成熟的语言，因此我们的Flink和Blink均基于此，之所以选择将SQL作为核心API，是因为其具有几个非常重要的特点，一是SQL属于设定式语言，用户只要表达清楚需求即可，不需要了解具体做法;二是SQL可优化，内置多种查询优化器，这些查询优化器可为SQL翻译出最优执行计划;三是SQL易于理解，不同行业和领域的人都懂;四是SQL非常稳定，在数据库30多年的历史中，SQL本身变化较少，非常稳定。当我们升级或替换引擎时，用户是无感知的且完全兼容;最后，SQL经过优化可以统一流和批。　　过去，我们既需要批模式跑全量数据，也需要流模式实时跑增量数据，因此需要同时维护两个引擎，并且保持两份代码之间的同步。如果使用SQL，我们便可以一份代码同时跑在两个模式下，但SQL是为传统批处理设计的，并不能为流处理所用。SQL定义在表上，而不是流上。传统SQL处理的数据集比较有限，查询一次只返回一个结果。但是，流处理需要不断接收数据，不断对结果进行更新，并且查询也不会结束，这导致其需要对历史数据不断修正。所以，SQL的很多概念无法直接映射到流计算，这就是在流计算上定义SQL的难点。　　

为了在流计算上定义SQL，我们需要引入几个概念。既然批处理需要定义SQL表的概念，那在流计算上也需要表的概念，我们需要将传统静态表扩展成动态表，所谓动态表就是数据会随时间而不断变化的表。此时，我们发现流和动态表之间有一种对偶性，也就是说流和动态表可以相互转换。将流的每条数据插入到数据库中，就得到了一张表;同时我们可以抽取动态表的changelog还原原始流。

从流计算到SQL，我们可以把它看成是连续查询。连续查询区别于传统的批处理查询，需要源源不断地接收数据，每收到一条新数据就会更新结果且结果也是一张动态表，那结果的动态表又可以作为下一个查询的输入，从而串起整个流计算。　　基于上述两个概念，我们可以在SQL上定义流计算。但是，流计算中的数据需要不断修正和更新，因此这些数据下发后可能导致最终结果的错误，我们需要把这些错误数据进行修正，这就涉及到流计算中一个非常重要的概念——Retraction。

为了解释此概念，我们举一个简单的例子，上图所示有一个点击输入流，它具备两个字段：user和url，经过第一个查询根据用户进行分组，统计每个用户的点击次数;进入第二个查询，根据点击次数进行分组，统计每个次数的具体点击人数。最终，我们会收到两条记录，点击次数所对应的人数。从结果明显可以看出计算有误，Mary的数据并没有合并计数，这就需要引入修正的概念。

如上图所示，经过修正之后，经过第二个查询时，Mary的总查询次数会被合并计算，Mary 1的结果会被告知撤回，从而输出正确的结果，这就是引入Retraction的作用。在整个过程中，是否触发Retraction以及发送方式均由优化器决定，用户对整个过程是无感知的。　　在此基础上，我们发现世界不需要所谓的Stream SQL语法，标准的ANSI SQL就可用来定义流计算，Flink SQL就是标准的ANSI SQL语法。其部分核心功能如下：DDL用来定义数据源表、数据结构表;UDF、UDTF、UDAF用户自定义函数，可以定制化用户复杂的业务需求;JOIN是一个比较复杂的功能，包括流与流之间的Join，流与表之间的Join以及Windows Join等;聚合功能包括类似Group AGG，Windoes Agg以及Over Agg等。

接下来我会结合实例对核心功能进行介绍。首先是装载数据，需要create table语法。如上图所示，我们先定义一张clicks表，然后定义表的schema、user、cTime以及url，with里是表的一系列属性，它是一个来自kafka的日志表，我们可以用SELECT * FROM clicks查询转载表里面的数据。

如果要将上述查询数据写到某个表中，我们需要用create table定义结果表，语法同上，创建一张 last_clicks 结果表，主键是user，通过INSERT INTO 语法将上述查询数据插入Mysql表中。

如果想把中间处理结果同时写入多个存储，比如把数据处理结果同时写到Mysql和HBase，如上使用CREATE VIEW 定义一个来自淘宝的点击记录，同时连续写多个INSERT INTO到Mysql和HBase。

接下来是Group Aggregate，也就是无限流量聚合。所谓无限流量聚合指从历史开始到现在的所有用户点击数据，如上查询展示的是根据用户分组，然后统计点击次数。如果来了一条Mary1的数据，我们就先插入该数据，后续如果Mary再次进行点击，我们就在原数据基础上进行修改更新，以此类推。

Window Aggregate是定义在窗口上的聚合，有别于上述无限流聚合，它的原理是是每个窗口对应输出一个结果，比如每小时每个用户的点击次数，需要在group by的结果上加上endT数据，也就是窗口标识。　　

接下来介绍双流join，目前我们支持INNER, LEFT, RIGHT, FULL, SEMI, ANTI等Join类型，举例说明双流Join的主要使用场景，比如把主流打成宽表，并补上额外字段等。如上图所示，我们需要将订单和物流表信息进行Join操作，在Join的物理实现上会有两份状态，用来存储两条流到目前为止收到的所有历史数据,淘汰机制时间设定为一天半一次。两者中任何一方信息延迟都会先在表中等待，直到同一个订单的信息与物流关联之后才会通过Join输出。

维表Join与双流Join类似，目前支持INNER, LEFT两种交易类型。维表Join的使用同样为补全主流，但想补全的字段在另一维表中。如上图所示，使用时首先需要通过CREATE TABLE 语法定义一张维表，此处定义的是 Products 表，存储与产品相关信息，查询同样使用Join语法。Order与Products表通过Products ID实现Join。关键字PERIOD FOR SYSTEM_TIME 是 SQL 2.11标准里的语法，意思是当前关联的Products是当前时刻的信息，关联之后不再更新信息。上图右侧展示的是维表Join物理执行的概念。我们可以根据Order去Products数据库里查询信息，最终Products维表返回关联信息。　　核心功能如上所述，接下来主要聊优化。维表中，订单O1查询时是堵塞等待IO的状态，此时无论如何调优性能，吞吐量和CPU使用率都上不去，因此我们引入异步IO功能。

如上左半部分为未引入异步IO时的状态，如上右半部分为引入后，此时若发起A请求，不需等待IO就可立刻发起BCD查询请求，然后异步等待返回结果。返回ABCD以后再管理输出，极大地提高了整体性能。

如上，异步IO使用时与维表Join只有一行配置改动，对于用户来说，这个使用是非常简便的。

第二个优化是大数据中的常见场景——数据倾斜。如上为改进之前，红色聚合节点出现数据积压现象，而紫色节点相对较空。

如果持续一段时间，红色聚合节点就会被打满，从而变为热点，所有上游map节点就会反压，停止处理数据进入等待状态，而下游的紫色节点基本处于空闲状态。

我们引入Local-Global 聚合优化。左图是未优化拓扑图，右边是引入Local-Global优化后的图，我们在Map后引入Local Agg节点，Map与Local Agg是链在一起的一个线程，之间的数据传输没有任何网络开销。Local Agg可以将收到的数据按照 key进行预聚合，然后将结果按照 key分发给下游Global Agg进行汇总。　　假如每个Map的 TPS 是每秒1万的数据量，全局就2个 key：红色和紫色。如果 Local Agg聚合的间隔是每秒钟一次，那么每个Local Agg能将1万条数据预聚合成最多2条(全局共2个 key)。那么Global Agg每秒钟最多收到只会三条消息，能有效降低Global Agg 的热点。优化后，我们对此进行性能测试，发现Local-Global 可以带来超过20倍的性能提升。因此，整个方案是十分有效的。