大数据平台架构方法论、模型与实践深度观察

近日笔者对大数据平台架构做了广泛的调研，目的在于对当前的主流大数据架构的方法论、模型与实践有一个全面的观察，以便为电商、出行、制造业、金融、媒体、智慧城市、物流、车联网等各行业朋友在理论与实践上提供一点参考，尤其以实时流式计算为核心。

1 方法论

目前大数据架构主流的方法论有三种：

• 2014年1月份Tweeter的Nathan Martz提出的Lambda
  

• 2014年7月份Linkedin的Jay Kreps提出的 Kappa
  

• 2015年Google的Tyler Akidau 等提出的Dataflow Model
  

这里不转述它们的内容，或做什么比较，网上很容易找到相关内容，个人建议最好看链接指向的原文。这里想特别指出的是，截止本文的写作，Dataflow 是最流行的模型，它不仅激励了 Apache Flink 的发展，而且 Dataflow 模型的作者 Tyler Akidau 等也是 Apache Beam 的 Founder 与 Commiter。Flink 和 Beam 都实现了 Dataflow 模型的很多概念，它可以广泛地应用于数据处理领域，包括 ETL、数据处理的服务等，不限于批流统一。

总的来说，大数据架构的整个发展历程可以简单地这样总结：首先，Lambda 批流并举、既独立又互补；然后，Kappa 倡导统一为流；现在，Dataflow 统一流、批、微批，并建立概念模型。不过 Kappa 只是提出了想法，而 Dataflow 将模型做了细化，包含了一些重要的实践探索与概念（Operator/Window/Trigger/Accumulate/Retract/Watermark/Session），并做了具体展开。

大数据架构发展到今天，批流矛盾早在前几年就显现了，各大互联网公司也早就遇到了这一矛盾。Google 提出 Dataflow 并不是空穴来风，它也有这种矛盾，分立的流计算 MillWheel 和批计算 FlumeJava，其实在发布 Dataflow 模型之前，Google 内部已然碰到了二者的矛盾问题及部分的解决。

Beam 最早在 2014 年作为 Google Cloud DataFlow SDK 发布，在 2016 年捐给了 ASF。Flink 和 Beam 都得到了三个有关论文的启发（MapReduce，MillWheel，DataFlow）。但是 Beam 并不是独立的 Runtime，它是通过 Runner 来与各种执行引擎集成的，包括：Apache Flink、Apache Spark、Apache Samza、Hazelcast Jet、Google Cloud Dataflow 等。Beam 支持各种语言，如：Java、Python、Go、SQL，而且还支持它们的 ML 库，如：Numpy、Pandas、Tensorflow。另外，Beam 以统一的 API 来处理流与批，而 Flink 是不同的 API（DataSet 和 DataStream）。有些大公司为平台支持语言少或者与机器学习框架集成问题而烦恼，也许 Beam 是不错的探索选项，Flink 官方鼓励 Beam+Flink 模式。

目前，虽然 ASF 的 Flink、Samza、Beam 都拥抱批流统一的理念，但其中，Flink 实现得最为充分、社区也最为繁荣。

2 Ingestion

2.1 模型及原则

总要有数据才涉及到对数据的处理，所以从数据的源发地开始看看数据的摄入情况，其基本的概念模型如下：

对 Figure 1 中模型的要求：

Ingestion 的原则：

• 高可用、高吞吐量；

• 限流或Backlog存储，也许Ingestion层本身资源不足、也许下游处理不过来；

• Connector，可自行扩展，支持多种协议与技术；

• 有丰富的Built-in Connector，在Ingestion环节尤其看中流存储；

• 容量可自动伸缩，Scalability；

• 支持主流集群管理，如：Mesos/Yarn/Kubernetes；

• 轻量的从Source到Sink的计算，完全可配置，计算不需要Shuffle；

• 不丢数据，Loss Tolerant；

• 保证送达，Guaranteed Delivery；

• 有全面的监控、告警；

• Auto failover；

Ingestion 管理的原则：

• 强大的UI；

• 动态配置；

• 策略配置，包括资源调度；

• 安全管控的配置；

• 发布的版本控制与审计；

• 集成元数据管理；

• 身份认证、授权、多租户；

2.2 技术推荐

基于上述模型与要求，以下按数据的源头、实际上也是对应的 Ingestion 方式，分别进行分析与技术推荐。

Web page/Mobile App/MES/IoT/System Built-in Producer， 都是从公网或内网发过来的情况，是被动接收。对于这类，可能很多公司都自研了，个人倒是建议采用 Apache NiFi，虽然国内有 DataX，但是 NiFi 的能力要丰富得多。

简单来讲，可以把它想象为一个桥梁，在企业级的全景图中，如果想让数据从 A 流到 B，并在中间做些转换可以考虑，它是系统间的自动化的数据加工管道，不需要编程，可以在可视化的环境中拖拽完成。它不仅可以连接各种主流大数据框架，而且支持基于各种协议的侦听、读、写，甚至 IoT 的 MQTT；它也支持对各种文件格式的处理，如：Parquet、CSV、JSON、Avro、ORC 等，还支持各种脚本语言。

这里之所以推荐 NiFi，是因为它已经相当成熟，有了每天处理几十亿数据的案例，虽然它也有算子构成的 DAG，但是更侧重两端。多年以来它在流管理、柔性伸缩、安全、易用性、监控等方面都在不断发展，它最突出的一点是声称 Loss Tolerant、Guaranteed Delivery，通过 WAL 和可插拔的内容库。NiFi 实在是多面手，如果要在企业系统全景图中的构件之间架设一个轻量流式处理的桥梁，值得考虑。它可以承担在公网上大吞吐量地接收消息的角色。虽然单纯从功能上看，自研并不难，但是要综合考虑到管理、安全、运维、监控、数据不丢失、自动故障转移、多租户等各个方面，需要做的工作还是很多的，所以作为企业级大图景在云中的最外层构件，还是推荐这种已经有相当积累的框架。有些公司不仅有侵入式的埋点，还有非侵入式的 Javaagent、APM、Pinpoint 等，如果要直接通过网络发送数据，也不妨尝试用 NiFi 来接收，如果落为文件也可以用，下面会提到。

Database，首先可以是定期增量或全量批拉的方式，这里主要讲实时的 CDC 方式，数据是基于 CUD 事件被推过来的。如果是 MySQL，可以采用 Canal 或者用 MySQL-Binlog-Connector-Java 自己写。前者有个不好的地方在于，它是分 Server/Client 的，用后者自己写可以把这个逻辑写成一个定制的 NiFi Processor，然后也许需要对代码逻辑做一些中间处理或者用 NiFi 界面构造 DAG，最后流入 Kafka，也可以同时流入其它存储。这种方式的实时性非常高，可立即对 MySQL 中的 CUD 事件做出反应，两种方式都需要独特的数据库权限。

如果是 Oracle，也要求实时性，那么 Oracle GoldenGate for Big Data 是一个选项，可以把 Oracle 中的交易变化实时地反映到各种大数据存储中，包括 Kafka。如果是 PostgreSQL，也有 CDC，不过 Oracle 和 PostgreSQL 都没有内置支持。

如果 HBase 在一线业务应用中直接作为存储，可以考虑写一个基于 HBaseEndpoint 的 CDC（Change Data Capture）。如果是 Cassandra 作为一线业务存储，官方有 CDC 支持。业务数据库的 CDC 环节一定要加上监控、告警的机制，实际上这两种机制要在整个企业 IT 大图景中所有的地方都加上，才能随时感知应该知道的系统的健康状况。对于 RDBMS，如果可以接受批计算，当然传统的 Sqoop 是一个选项。

System log，可以用 Apache NiFi 中的 TailFile（支持 Rolling）这个 Processor，中间加上自己写或拖拽处需要的处理逻辑，最后吐到 Kafka 等存储中。或者，使用经典的 Flume 也是没毛病的，但是简单就是美，都用 NiFi 岂不是更好。

这么看来，其实用 NiFi 一个框架就可以解决所有数据的问题了，而且，NiFi 具备综合的管理能力，如：跟踪、监控、拖拽开发、多租户、HA 方案等。可以把 NiFi 看成是万能的 Integrator，但不要把复杂的计算逻辑让它来担当，毕竟那不是它的强项。有趣的一点是，它还有 Backpressure 的能力，这意味着在整个企业大数据架构的最外延、数据的入口处可以有一个控流阀来控制流速，也许在某种场景是个不错的强项。

3 元数据管理

元数据概念起源于 20 世纪下半叶，广为传知是在 RDBMS 时代。而到了当今这个数据时代，应该用更广袤的视野来看待它，包括：

• 传统的RDBMS的元信息，库、表、字段、关系、UDF、触发器、存储过程等；

• 关于计算体的元信息，DC、集群、产品、应用、系统、作业、服务、UDF、算子；

• 关于人的元信息，地域、组织架构、人员、流程；

• 关于网络的元信息，如：跨交换机是有距离的，延迟是不一样的，这部分也应当为计算所知；

• 关于业务的元信息，各种业务术语、过程、领域、BU、渠道等；

从价值的维度看，目前数据已经被提升到资产的高度。既然数据是资产，对它的管理就应当向财务方向看起。现代的元数据管理是构建完善高效的企业 IT 管理、治理体系的基石。在企业范围内，可以这样比喻，如果把以上所有大数据体系构成要素统一抽象为大数据王国的公民，如果没有元数据管理，就相当于这些公民之间没有共同的语言，自然就会有歧义、有纷争、有困惑、有乱象、有翻译、有成本。其中人、组织的要素也非常重要，例如：资源的交接、权限、计费、计量、资源回收、预算、流程等都涉及。企业 Digital Twin 的运行也离不开元数据管理。车同轨、书同文、行同伦，天下大治期矣。

本文之所以这么早进入元数据章节，是认为它发挥作用应该越早越好，所以就放在了 Ingestion 之后，其实在 Ingestion 阶段，就应当做些必要的元数据方面的工作了。在数据进入整个企业大数据体系边界，就应当把该做的事情做了，不论是否为结构化数据，至少要先登记造册，哪怕是扔到数据湖里，如：

• 什么时候到的；

• 属于哪个业务领域、哪个部门的数据；

• 来自哪个系统、哪个应用；

• 基本度量；

• Schema；

• Format；

元数据管理涉及到数据治理、Lineage、监控、作业、业务信息、访问控制、审计、合规、可视化、数据标准、数据质量、版本控制、计算作业、发布管理、计量计费、告警等方方面面。它关系到企业如何有序地、高效地、稳定地、弹性地、智能地运行。虽然 Haddop 生态基本上用 HCatalog 来存储元数据，但仅仅存储是不够的，对于整个企业而言是非常局限的。一个字段名及其语义标签在一个企业中存在多种表达，而内容本质上却是一个，这不是什么稀奇的事。

在企业数据的大图景中，远不止纯粹数据层面的元数据信息需要整合管理，还包括各种其它对象，如 UDF（Kylin、Hive、Flink、Spark、Storm、Phoenix 等）、作业（Flink、Storm、Spark 等）、部门、人员、数据中心、集群、业务要素等，这样才能更充分、高效、系统、完整地保障企业的运行效率。对于作业，更理想的开发和使用方式应当是配置的、声明式的，而剩下的事情交由平台完成，这样才能使用户以更加一致的、统一的、低成本的、可追溯的方式来执行所需的计算任务，而这离不开针对作业的元数据管理。因此，有一个独立的、统一的元数据管理平台是上策。

3.1 技术推荐

Apache Altas 也许是一个不错的起步选项，它是针对 Hadoop 生态的元数据管理工具，以基于 HBase 的 GranusGraph 作为存储，Solr 用作搜索，用 Hook 感知 Hive/Kafka/HBase/Storm/Sqoop 的元数据的变化，遗憾的是没有 Flink 和 Pulsar，不过可以定制。Altas 支持 HA。

Atlas 用 Type/Entity 模型来组织所有的元数据对象，它们的关系相当于 OOP 中对应的 Class/Instance。Type 可以分为多个 Metatype：Metatype/Enum/Collection（Array,Map）/Composite（Entity, Struct, Classification, Relationship），Composite 可以有多 Attribute，而 Attribute 可以指向 Metatype 从而建立丰富的关系，有趣的是 Entity 和 Classification 是可以继承关系的，真正存放元数据信息的叫 Entity，例如：一张 Hive 表。Atlas 可以从 Kafka 收到关于 Entity 和 Classification 上的 CUD 事件。

Atlas 还提供了 REST API 来从 Atalas 查询和搜索 Entity、Lineage、关系、Type 信息，也可以对元数据做 CUD 操作，某种场合下这在各种流计算的算子中是非常有用的。Atlas 用 JanusGraph 管理丰富的元数据对象之间的关系。

业务元数据这样组织的，核心叫 Term，一个 Term 可以属于多个 Category，而 Category 是 Hierarchical 的，一个 Term 还可以关联多个 Entity，一个 Term 还可以有多个 Classification，Term 之间还可以建立丰富的关系，如：反义词、同义词等。Glossary 是颗粒度最大的，它下面有 Term 和 Category。

安全方面，支持双向 SSL、SPNEGO-based HTTP（Simple/Kerberos）、JAAS（Kafka 需要）；Authentication 支持 LDAP、Kerberos 等几种方法；Authorization，可以针对在 Metatype 上的 CUD 操作、Entity 上的 CRUD 操作，可以插件方式定制自己的授权组件，除了默认的 Simple Authorizer，还可以用 Apache Ranger 来作为 Authorizer，Ranger 是为 Hadoop 生态提供的综合的数据安全管理框架，突出裨益在于集中管理、审计用户的访问行为和授权管理行为，其模型为 Role/Attribute Based Access Control，支持各种策略。

4 流存储

4.1 模型及关注点

Dataflow 模型需要一个相应的存储系统来支撑，流存储的基本概念模型如下：

流存储的核心能力是分布式的、流式的高速读写，这是流计算在存储上的必然要求。其它特征要求就不费周章了，也基本上名释其义，粗略理解还是可以的，可以结合自身的生产实践来从架构上作参考。

从实践的角度看，不管具体采用何种技术甚至自研，应当注重以下方面的投入：

容量的伸缩（Scalability）

好多大公司都在这方面遇到过挑战。数据如何做到整体分布尽可能均衡，一个 topic 还罢了，很多个怎么办？最好有一个全局范围的、不影响读写的、自动的 Rebalance。不论方式如何，有一个原则不能变，一个存储文件中不应当存在 1 个以上的 Topic，否则顺序读写就不成立了，而这正是流存储必须坚持的，这和 HBase 不一样。那么在此原则下，随着时间的流式，存在 Data Skew 是必然的，因此，Rebalance 也是必然的。流存储本身最初也不可能预知每个 Topic 将如何增长数据，当然加上预测能力自动 Rebalance 是个好主意。

不仅数据会 Skew，内存的使用也可能有 Skew 的情况，这也是节点宕机、或出现其它问题的原因之一，同样需要在不影响读写的情况下自动 Rebalance，而且不影响副本数量。

监控与告警

这是所有系统必须做到的，流存储作为 Dataflow 模型计算体系的必要组成部分，是数据的大 Buffer、大 Hub，它有了问题，影响可想而知。包括 CPU、Disk、Memory、Network、Producer、Consumer 等都要监控，一旦发生情况，就应当直接通过告警机制连通 On-call Shift 及时处置或者 Auto Heal。不论采用什么样的流存储技术，都严重依赖内存大小、磁盘吞吐率、网络延迟、文件系统调优，因为流存储的重点就在于分布式环境下流式的高速读写，所以必须把这些方面的监控做好。

磁盘用量的管控

要综合根据所有的 Retention（Time/Size Based）、复制的情况对总体磁盘用量有个预判，现有的磁盘容量在将来是否够用，要在使用的过程中动态判断，所以辅助一些 ML 手段去预测还是非常好的。如果用 Kafka，在计算中，要记得它是以 Segment 为单位进行删除，在 Retention 到限以后。

监控消费滞后

如果消费速度小于生产者的写入速度，意味着终端用户看到的是不新鲜的数据、不及时的数据，即：新鲜度不好、不及时。其实监控读、写滞后，都有意义，读 Lag 指读的 Offset 所对应的时间点与当时写的 Offset 对应的时间点之间的时间差，写滞后指与上一次写动作之间的时间差，它们反应的问题是不同的。也许技术上没问题，但是业务上一定会对数据的新鲜度有要求。

4.2 技术推荐

当前，大家首选 Kafka 的可能性比较大，它凭借对 Pagecache 的顺序读写和 Zero-Copy 完成高速读写，这非常棒，从模型角度看高速读写也应当是以 Dataflow 模型为核心的批流统一架构体系中的流存储所必须的能力。这里就不赘述它优秀特征，只对大家在实践中可能遇到的问题做个提示：

• Topic达到几千个可能就会有问题，主要是由于file handler的限制；

• 租户隔离，当一个节点上的一个分区在一个消费者或一个生产者出现滞后的现象，那这个节点上所有其它分区上的动作都会滞后；

• 缺乏针对多租户的计费计量能力，在生产者和消费者上；

• 队列能力支持不够，如：Delay Queue；

• 监控方面做得不大好，没有完整的监控工具；

• 有时写生产者和消费者时，可能希望同时使用多个Topic，如果对Topic有通配符能力可能更灵活；

• API是由不同的个人和公司维护的，与整体发展不同步；

• Likedin使用Kafka非常重，达到7 Trillion Per Day，4000 Brokers。为此，Linkedin内部做了补丁分支（不是Fork），为了满足增减Broker的运维工作和特有的特征，已开放到Github。Scalability方面的主要问题是，内存造成的Controller Failure和Slow，解决的方法是重用UpdateMetadataRequest。还有一个问题是启动或关闭一个Broker（一次只能操作一个）比较慢，改进方法是减少锁竞争。由于运维的需要，增减Broker会很频繁，Linkedin给Broker增加了维护模式，以免有消息不断流入，这样才能安全地移除，最终保障副本的数量。

• 沃尔玛也碰到了恢复失败节点而造成不好的影响，缺乏再分区的能力，如何跟踪Rebalance的进度，并确保期间生产与消费都正常还是满重要的；

• 缺乏Cloud/Rack/Azure Awareness；

Kafka 是大部分人首先想到的，毕竟早 5 年起家，在社区的丰富性和规模上都有优势。而作为后起之秀的 Pulsar 的社区在 Slack 上很活跃、在壮大、响应快，这里把 Apache Pulsar 的特点介绍一下，在生产实践中可能不失为一个可以考虑的选项（当然，任何技术在生产实践中都可能碰到问题）：

• 经历过每天处理上千亿的数据、上百万的Topic的生产考验，Battle-Tested；

• 流与队列的合体，单一的API，而Kafka侧重流；

• 可以通过Producer或者Broker直接去重；

• 分区不是必须的，不必像Kafka一样要考虑分区与Consumer的数量。它在Topic与Consumer之间加入了Subscription（Exclusive/Failover/Shared/Key_shared）抽象，甚至于一个Subscription可以对应多个Topic。分区时吞吐量可以非常大。

• 资料表明，Kafka加一个Broker，再平衡分区会比较慢，而Pulsar把Segments分布式地用Bookeeper存储缓解了这个问题。这道理是讲得通的，因为在kafka中数据的存储逻辑细分下来依次是这样的一对多的关系：Topic/Broker/Partition/Segement/三个文件，因此，在Kafka中一个Partition的所有Segement都在一个Broker上，而在Pulsar中是分散到Bookeeper；Pulsar的负载均衡是自动的，它检测完全无状态的Broker的Cpu/Mem/Network IO，自动执行Rebalance；

• 支持Geo-replication，即：跨机房复制；

• 阿里、滴滴等大公司支持的行业分析公司的基准测试表明，Pulsar比Kafka在吞吐量和延迟方面都有显著的优势；

• 支持多租户，每个租户可以有多个Namespace，在加上Quota、访问控制、Rate-limit，可以实现丰富的权限管理。通过Namespace可以实现集群之间的复制；

• 以Quorum的方式副本复制，性能表现更为稳定；

• 分层存储，Tiered Storage，允许把老的积压数据卸载到长期的、经济的存储中，如：HDFS。例如：用户行为数据需要长期保存，以便机器学习训练用于训练模型；

• 生产者到消费者的消息加密，保障绝对的安全，Pulsar自己不保管Key；

• 一个集群可以服务上百万的Topic，但是Kafka达到几千个就出状况了；

• 现成的CDC Connector for MySQL/PostgreSQL，还有好多内置的Source/Sink Connector；

• 支持消息的TTL；

5 流计算及结果的应用

5.1 主流技术参考与架构

5.2 Flink 及其生产实践

目前的流计算实现，Flink 几乎是不二之选，好多公司在做从 Sorm 和 Spark 到 Flink 的迁移，而且它符合 Dataflow 模型，很多大企业都有成功的运用，社区发展得如火如荼，改进也很迅速。虽然说 Beam 出自 Google 本家，但它的定位不一样，可以认为它是 Spark/Flink/Samza 等流计算引擎之上的一层东西，它为定义和执行数据处理流程提供了不同语言的 SDK，为不同的引擎提供了不同的 Runner。这里不介绍 Flink 特征，官网最好了，仅对生产实践可能有帮助的内容做一些分享：

• Latency与Guaranteed Delivery之间的权衡，对于Mission-Critical的作业，保存状态很重要，以备某种状况下的接续恢复执行、甚至倒带。然而，对于一个算子，如果采用Exactly Once，只要它的算子存在多进或者多出的情况，就要做Barrier Alignment，即：要等待第n个Checkpoint的数据都到达，否则不会处理已经到达的第n+1个Checkpoint的数据，这当然就增加Latency；如果采用At Least Once，则不发生Barrier Alignment。所以要根据具体业务情况决策，另一个途径是降低Parallelism，如果可以的话。

• 存储状态造成的Latency，三种State Backend默认都是以异步方式保存状态而不阻塞流的处理，但是Heap+RocksDB的组合方式是不支持Timer State的；

• 资源隔离，Yarn+Cgroup隔离CPU和Memory；用Yarn Node Label实现机器、容器颗粒度的资源保障优先级和专用保障；

• 使用Asynchronous I/O 调用dubbo接口，Zuul、DB、HBase等外部接口；

• Blink Planner有更好的性能表现；

• App1（由多个作业构成）和App2可以共享中间结果，结果共享可以很大程度上节约资源；

• 如果对一个流要做事件处理，但是可能有好多规则、不仅规则本身会变化、而且规则也会增减，配置驱动是最好的方式；

• 宜早不宜迟

• Java对象的空间代价差不多是翻倍的，相对于Primitive Type而言。Flink在网络计算、Checkpioint的时候是要涉及Serde（序列化、反序列化）的，所以在Serde上发力肯定会在内存优化、速度上大有收益的，例如：自己造一个时空高效的、完全基于Primitive 的数据结构，也是完全可行的；在什么阶段实施这个要看情况了，因为个性的Serde可能意味着与周边的第三方系统不兼容，但时空收益那是肯定的；

• Ingestion阶段，可以把一个消息按不同的分区方式（即：用不同的Key值打散）产生多个流输入流存储中，这样可以以空间换时间，因为它可以让流计算阶段省却网络Shuffle。

• Ingestion阶段，第一时间维护维表，Redis/HBase/es/Aerospike，如果采用外在存储方式保存维表的话；

• Ingestion阶段，第一时间把ID变成数值型，可以用bitmap，在有些问题上很有用；

• Pushdown，不论是使用API还是SQL，都要尽可能把聚合、过滤等计算先在小数据集上完成、然后再Join，而不是在Join后的更大的数据集上进行。能在一个Task中先做到的就在一个Task里先做尽、能在一个Taskmanager中先做到的就在一个Taskmanager里先做尽、能在一个Table中先做到的就在一个Table里先做尽、最后才是Join等需要网络Shuffle的动作，这是高效计算的次第；

• Ingestion阶段，分而治之，在使用上完全不相干的数据就不应当发到同一个topic；

• 数据仓库

• 实时数仓中，需要用到维表，方法估计至少5种，其中以Temporal Table Function方式最佳，数据量大、实时性强；

• 离线数仓，数据表明基于Flink的效率至少比HiveOnTez要高；

• 流存储的分区数是Flink Parallelism的整数倍为好，以防数据Skew；

• 常见问题

• 数据Skew、频繁GC、对周边慢读慢些、大窗口、Serde（Shuffle and State Backend）；

• 注意设置State TTL，以免长时间跨度的计算需要数据的时候数据已经被clean了；

• 监控不足，作业级、算子级、Backpressure、Executionstate、Jobstatus、ZK、上下游；

• 作业发布

• 检查与在运行的老版本的state的兼容性问题，回滚能力，选择Sheckpoint、批量升级、审批流程；动态修改作业的执行逻辑是有风险的，如：SQL、代码，可能需要增强State的维护逻辑（如：状态比对）、甚至使用其它存储；State Processor API挺有用的，可以读写分析State，所以对State维护很有帮助；

• 作业可以是基于API写的，但是更推荐可配置SQL的方式、并结合版本控制。一种是把SQL翻译成DataSet API或者DataStream API；一种是把自定义的某种声明式的配置翻译成API的执行；还可以通过把SQL提交给AthenaX来提交作业；要么做自己的SQL执行机制，Query Parser/Optimizer/Planner，然后改造运行时，跳过2个API，不过这就要改Flink了；

• 还有一种情况是，在不重新启动作业的情况下，动态更新SQL及其它配置也是可行的，结合ZK的Watch Push机制；

• 总之，SQL具有可维护性好、接受度高、维护成本低、可动态修改、规范、可配置等优点，一直是不错的User API，尽可能与具体实现技术解耦、保持Portability，是个不错的方向，也更容易为没有专业技术背景的Citizen Developer服务，而不是让终端用户学习各种技术；

• 如果想在流上应用ML模型，先把计算结果流入流存储（如：Kafka），然后加一层消费者作为模型应用层更为合适，由其是在应用多个模型在同一个流上的情况，这样可以把特征准备与模型运用解耦，互相不干扰、独立Scale，而不和流计算混合在同一个进程中；

• 复杂事件的处理

• CEP，SQL的match_recognize（ISO标准SQL的一部分）或者DataStream上的API，它的强大在于通过Pattern来识别是否一个事件发生了，例如：统计没有点击的广告展现（涉及展现和点击两个事件，但是在规定时间内点击事件没有到达），就可以用这个。Flink-siddhi是个值得参考的实现；

• Stateful Function framework，融合了Flink与FaaS（例如：AWS Lambda 和微软的Virtual Stateful Actor Model），这是个复杂的框架，如果描述概念恐怕不好理解，这里举个例子：假设任务是这样的，从Kafka输入的消息为Tuple（user_id=111，“张三”），作为结果向Kafka输出的消息为Tuple（user_id=111，“欢迎张三的第N次访问”），这里显然要维护每个人的访问次数，该框架的做法是，当一个Taskmanager收到一个用户的消息的时候，可以把这个消息按用户路由到维护着该用户状态（访问次数）的函数去处理，而这个函数可能位于当前的Taskmanager、也可能位于集群的另一个Taskmanager，就是这样的。再扩展一下，在网约车的场景里，司机与乘客发生两个消息流进入系统，通过司机、乘客、Geo Index、Ride等有状态函数之间的交互最后完成乘客的Book和司机的Bid。这个很像电信领域的霸主Erlang的Actor Model，其实Flink本身的协调也用到了Actor Model（即：Akka）。总结，这个框架一方面可以用于复杂事件的处理（状态机），另一方面也可以用于极低延迟要求的动态AI模型的运用（每一个待预测个体都有自己的状态），不过像金融领域的情况可能还是比较挑战，因为特征计算本身可能就需要高强度计算，涉及4、5千个变量不稀奇。

• 基本方针

• 作为用户的开发者的自由度越小越好，从总体的规范度、代码质量、系统稳定性、可维护性、可控性、综合成本等方面考虑，通过提供Scaffold、Framework、SDK、Platform、Configuration、Code Checker、UDF、SQL等方式渗透到软件研发的全生命周期是非常可取的，当然这与开发者的探索欲望是矛盾的，但从企业的角度是非常值得的。

• 要尽可减少网络Shuffle次数和IO的量、高效利用内存、索引、增量计算、少做无用功、复用累积成果、高效的Serde；

5.3 支持 SQL 的技术

大家非常希望数据具有较高的新鲜度，即：在企业范围内从数据产生的那一刻直到被最终真正用起来的那一刻之间的时间越短越好。基于这样的期望，自然想到直接对接 Kafka 的 DB，然后用 SQL 直接得到结果，毕竟 SQL 有很多优势，以下介绍几种选项：

• Rockset，它是由一群原来在Facebook做搜索的、在Google做Gmail的、在Oracle做数据库的人在一起做出来的，支持各种数据源，尤其是Amazon云上的，包括Kafka，它支持Full ANSI SQL，通过JDBC或REST在其上可以直接用Graphana/Tableau/RStudio/Superset/Jupyter/Redash/DataGrip，TB级的数据量、毫秒级的Latency，可见其强大，不过它不开源。

• Presto-Pinot-Connector，Presto擅长ANSI SQL Query，而Pinot擅长OLAP，其实Pinot和Rockset在本质的精华上是一样的，就是对所有的字段做倒排索引，Pinot也可以直接接Kafka，这个东西也不开源，是Uber做的。Pinot本身是开源的，有很强的的OLAP能力，2019年社区强烈建议增加对SQL的支持，如果有资源也许可以尝试利用Calcite做自己的Parser/Validation/Optimization/Plan然后对接Pinot。

• SQL-on-Kafka-with-Presto，一个Presto接多个Kafka集群，一个Kafka支持多个Topic（Table），这样就可以通过Presto对Kafka 使用SQL了，可以试试，没找到性能有关资料。不过应该都没有Rockset和presto-pinot快。

• Couchbase，虽然不是完全支持ANSI SQL，但它有自己的查询语言N1QL（SQL for JSON），N1QL兼顾了大家对SQL的熟悉与JSON的数据结构，其结构、语法都非常像SQL，虽然它没有Join，但是支持子查询，支持SDK和REST两种方式来提交SQL，还支持函数、窗口等好多特征，这些特点远胜于搜索引擎，但是索引要主动建立，这个不如搜索引擎。就SQL的支持程度而言，还可以考虑Druid，之所以和Couchbase放在一起，是因为它也支持子查询，它叫Semi-join，虽然不支持Join，但是Druid的优势在于它像搜索引擎一样不用有意识地建索引。

• ksqlDB，它的作者是Kafka的Co-creator，ksqlDB通过REST支持SQL，支持负载均衡和自动故障转移，支持Join。

• Clickhouse对SQL也有相当不错的支持。

• 还有一个办法是把数据持续流入Elasticsearch或者Solr，不过这两个的Join能力都较弱，如果不需要Join那还是不错的。

5.4 一个优秀的 K/V 存储

Aerospike 虽然不支持 Full SQL，但是真的是非常值得考虑的 HBase 的替换选项，尤其在要求更高速度的时候。Flink 中可以通过 TableFunction 实现对 Aerospike 的读写，Aerospike 的优势包括：

• 采用C语言开发；

• 读写明显比HBase快，K/V查找速度比Redis快20-30%、比Mongo快50%、比Cassandra快45%，非常适合实时性高的应用；

• Flash-optimized In-memory NoSQL（Key/Value），针对SSD做了专门的优化；

• 支持二级索引；

• 支持UDF；

• 本地集群的Immediate Consistency和跨数据中心的Eventual Consistency；

• 支持原子操作；

• 支持基于用户与角色的访问控制；

• 支持ACID操作；

• Auto-sharding & Auto-healing，扩容无需手工干预；

• 可以部署到Kubernetes；

• 开源，在Reliability、Latency、Scalability上都非常优异，广泛用在各个行业的关键业务的实时场景，如：金融、电讯、广告、游戏；

• HA简单；

• 用了非常复杂的哈希算法来避免Data Skew；

• 支持多种语言；

• 支持Geo Index；

• 和RDBMS相对应，其数据模型（括号中是关系型数据库的概念）是Database/Namespace(Tablespace)/Set(Table)/Record(Row)/Bin(Column)，但它是Schemaless。Record上可以有TTL，Generation相当于HBase的Version。一个Namespace最多可以有32k不同的Bin。