数据实时接入和分析面临的挑战

随着大数据时代的到来，企业在生产和经营活动中产生的各类数据正以前所未有的速度增长，通过对实时及历史数据的融合分析，及时挖掘业务洞察和辅助决策，已成为企业的普遍行动。

有一种观点认为大数据存在“3V”特性：Volume, Velocity, Variety。这三个“V”表明大数据的三方面特征：量大，实时和多样。这三个主要特征对数据链路系统的影响尤为突出。多种多样的数据源，海量的数据以及实时高效的采集传输是数据链路系统主要面对的几个问题。即系统需要在满足实时性指标的同时，也具备生产环境下的高可用性和易用性。

下图是一个非常经典的数据链路的架构图。从左到右，依次是数据源、数据接入层、数据缓冲层、数据分发层、数据目标，可以看出，搭建一个完整的数据链路非常复杂繁琐。

在云原生的浪潮下，企业需要聚焦业务，迫切需要简单易行，零代码地配置搭建起自己的数据链路系统。因此数据链路系统需要如下几个特征：SAAS化、低代码化、简单易用、稳定可靠、高性能、按量付费，以达到整体上的降本增效。

基于上述诉求，我们推出了数据接入平台（Data Import Platform）。

关于数据接入平台

腾讯云消息队列CKafka推出的数据接入平台（Data Import Platform），是腾讯云上SAAS化的数据接入和处理平台，协助客户方便快捷地完成一站式的数据接入、处理和分发。平台提供基于 HTTP/TCP 协议的 SDK 协助客户快速完成数据上报、基于 CDC（Change Data Capture）机制快速订阅、存储多款数据库（MySQL、PostgreSQL、MongoDB 等）变更信息，打通了多款云产品的日志投递。并提供了简单可配置的数据清洗 (ETL) 能力，以及丰富的数据流出渠道，协助客户低成本搭建数据流转链路，构建数据源和数据处理系统间的桥梁。

DIP类似于传统大数据解决方案中Kafka+Flink的角色，提供了通用的数据连接、处理、流转的功能。核心诉求是希望可以协助客户低成本的搭建整条的数据链路。根据二八原则，DIP希望解决大部分通用的数据连接场景。而对于业务属性强，逻辑复杂的还是需要依赖Flink等流式计算引擎来实现。

在离线计算场景，DIP提供了一个缓冲队列的作用，同时由于DIP提供了各种MQ与其他腾讯云上下游产品的对接功能，所以又扮演了数据分发枢纽的角色。

DIP是由腾讯云上CKafka孵化出的数据接入产品，底层基于开源Kafka Connector和自研接入分发层。从本质上来看，Kafka是消息队列，属于存储产品。而DIP是数据接入分发平台，定位为存储层（MQ）的上下游数据连接。

IP旨在围绕消息队列（MQ）生态，做好上下游的数据连接。消息队列属于PAAS产品，DIP定位为SAAS产品，提供一站式的数据链路搭建方案。DIP希望达到低代码、免运维、按量付费、Serverless化的效果，后续会支持多种消息队列协议接入。

仅需通过简单的界面配置，轻松完成数据上报、清洗，存储的链路搭建。屏蔽数据接入过程中底层复杂的系统搭建、组件运维过程。

支持云上、云下（跨云、混合云）场景、支持自建和云上服务的数据连接。实现了五大类(数据库、文件、MQ、主动上报、日志)，15+云上产品打通，一站式实现数据的接入和流动。

接入层、处理层、分发层均为分布式跨可用区部署，遇到故障即可自动切换，服务可用性不低于99.9%。

在数据采集、上报和流转整条链路过程中，实现秒级的接收、处理、并分发到下游系统。

数据接入平台支持不同租户间网络隔离，支持数据上报集成 CAM 鉴权、数据流转集成 SASL 权限控制，严格控制访问权限，保证数据安全。

无需预估业务容量，系统会根据流量规模自动弹性伸缩，保证波峰时系统可用性。按需使用，Serverless 化的完成数据接入、处理、转储的整个流程。

DIP应用场景

在系统开发中，会有一些数据统一上报的需求，比如监控数据、用户行为数据、APP操作数据等，需要将数据统一上报到服务端，供业务进行查询、处理、分析等。如手机 APP 的操作行为分析、前端页面的 Bug 日志上报、业务数据的上报等等。

一般情况下，这些上报的数据都需要转储到下游的存储分析系统里面进行处理（如 Elasticsearch，HDFS，数据湖等）。在常规操作中，我们需要搭建 Server、购买存储系统、并在中间自定义代码进行数据接收、处理、转储等。而server端需要考虑性能、稳定性、扩缩容等工作，实际工作量较大。

DIP 旨在用 SaaS 化的思路解决这个问题，目标是通过如下两步：界面配置、SDK 上报，完成整个链路的搭建，并基于 Serverless 理念，以按量计费，弹性伸缩，无需预估容量等方式，简化客户的研发投入成本和实际的使用成本。

申请一个接入点ID即可完成数据接入，DIP提供了两种协议的SDK，HTTP协议及kafka协议。通过HTTP协议可以将数据上报到Kafka/Pulsar或其他消息队列，可以屏蔽多种消息队列的复杂SDK使用。

DIP 支持基于 CDC（Change Data Capture）机制订阅多款数据库的变更数据。如订阅 MySQL 的 Binlog，MongoDB 的Change Stream，PostgreSQL 的行级的数据变更（Row-level Change），SQL Server 的行级的数据变更（Row-level Change）等。例如，在实际业务使用过程当中，业务经常需要订阅 MySQL 的 Binlog 日志，获取 MySQL 的变更记录（Insert、Update、Delete、DDL、DML 等），并针对这些数据进行对应的业务逻辑处理，如查询、故障恢复、分析等。

在默认情况下，客户往往需要自定义搭建基于CDC的订阅组件如（Canal、Debezium、Flink CDC 等）来完成对数据库的数据订阅。而在搭建、运维这些订阅组件的过程中，人力投入成本较高，需要搭建配套的监控体系，保证订阅组件的稳定运行。

基于此种情况，DIP 提供 SaaS 化的组件，通过界面配置化的完成数据的订阅、处理、转储等整个流程。

大多数情况下，数据接入后，需要进行简单的处理过滤和归一化，才能往下游进行下一步存储、查询和分析。简单的处理过滤和归一化就是数据的清洗与分发，数据清洗是指数据A变成数据B，数据分发就是指Kafka有一份数据既想分发到ES又想分发到COS，同时也希望计算平台可以计算。

例如，在简单的数据群里，在自建的场景中，对原始的日志进行格式化，进行分割重新组合、类型转换等这些操作，清洗完一个数据，再存到下一轮去。常规的做法就是需要写代码或者使用Logstash等开源组件进行ETL操作，但处理效率低，成本高，需大量配置才能处理海量数据，经常性遇到性能问题导致上有数据堆积。

如果使用DIP，我们会提供产品化配置界面，可以完成一键化配置，低代码实现数据的过滤和清洗。也无需关注扩缩容的问题，省去开源组件的学习成本，同时减少组件运维成本，支持用函数进行高级清洗，我们旨在打造一个更SAAS化的产品。

在实际业务过程中，用户经常需要将多个数据源的数据汇总到消息队列中，比如业务客户端数据、业务 DB数据、业务的运行日志数据汇总到消息队列中进行分析处理。正常情况下，需要先将这些数据进行清洗格式化后，再做统一的转储、分析或处理，创建整个数据链路就比较长。

DIP 支持将不同环境（腾讯公有云、用户自建 IDC、跨云、混合云等）的不同数据源（数据库、中间件、日志、应用系统等）的数据集成到公有云的消息队列服务中，以便进行数据的处理和分发。

DIP 提供了数据聚合、存储、处理、转储的能力。简而言之，就是提供数据连接集成的能力，将不同的数据源连接到下游的数据目标中，这样搭建数据链路就比较方便。创建任务后，整个任务的运行状态都是完全透明的，比如全链路监控、数据审计等，以保证数据在数据链路中不会丢失。

在整个数据链路搭建中，客户可以感知到的成本几乎为零，在这个基础上客户还可以享受到其他的服务，我们采用的是以serverless为底层的机制，自动扩缩容按量付费，既节省人工成本，又节省经济成本。

DIP技术架构

DIP的数据架构分为两层，一层是数据面，一层是管控面。管控面是指资源管理、监控调度，自动扩缩容、迁移等管控的操作。比如监控采集，客户可以把整个任务各个环节进行监控采集，配置告警。

DIP的核心还是数据面的，下图是DIP系统架构图：可以看到根据数据流转的链路，分为数据接入、数据处理、数据流出三大块。

数据接入的方式有三种：主动订阅、数据上报、自建IDC到混合云、跨云或公有云等多种云场景下获取数据。整个数据层面是多个引擎运行的。

在数据接入模块，DIP支持的数据源又可以分为三大类。一类是像mongodb、mysql这样的服务类数据，第二类是通过我们维护的日志采集器，采集到的TKE、CVM等等的日志数据；第三类是HTTP上报的、WEB、APP产生的数据。前两类数据源我们都会提供完全产品化的配置界面，在控制台简单点击配置就可以连通，用户不需要关心底层实现。第三类主动上报类的我们会提供SDK和接入点，把接入点写入到SDK就可以实现数据上报。

上面提到的异构数据源接入之后，我们提供数据清洗归一化的能力。对于通用的清洗模板，我们会提供完全产品化的配置界面，对于一些高级的自定义清洗规则，我们支持使用灵活的函数计算。

通过数据接入和实时计算，可以把多种数据源的异构数据实时清洗成统一的结构化数据投递到下游es、clickhouse等多种大数据系统或者存储系统。投递过程也是完全产品化的配置界面，只需要选择数据源和数据目标，点点点即可。

对这些数据流转任务，支持查看监控，可以查询在任务中投递的消息。之后DIP会支持任务编排，在控制台上可以清晰地看到数据流转链路，也可以对这个链路进行一些像加入和删除节点、编辑和查看监控的操作，同时支持指定schema，指定消息数据的格式。

通过上面这几大模块，我们就可以实时接入APP、WEB、IoT和数据库等产生的异构数据，统一管理，并投递到下游的分析、归档等系统，构建清晰的数据流，更好地释放数据的价值。

Connector可以简单理解为是一个依托于Kafka的分布式任务管理器，JDBC等不同的插件以线程的方式运行在整个分布式任务引擎中。

比如当创建任务的时候，如果指定三个并行度，Connector就会创建三个线程，它的任务就会调度到集群的每台机器上运行。Sink也适用于这套机制，Connector本身支持很多插件（如HDFS、HBase、S3、ES等）。这些插件就会在集群中运行，自动调度，会注册很多监控指标。我们的监控平台就会采集这些指标，依托这些指标去告警，扩缩容等，保证整个集群的稳定性，比如负载是否达到了一个阈值，当前的资源是否需要扩容，当前的任务是否存在异常等。

数据流引擎支持HTTP和HTTPS的协议接入，我们提供了各个语言丰富的SDK，当客户拿到SDK后，简单来说新创建一个对象，发送数据就可以了，数据就会上报到服务端进行存储。

服务端有几层，首先接收层分布式部署，可以水平扩容，每台机器节点都会维护底层和Kafka缓冲的连接池，在连接池中接收到的数据会直接转发存储到消息缓冲层中。数据流虽然从流程上看比较简单，但其实需要考虑很多情况，比如是不是有负载倾斜，是不是有一些请求需要有一批独占的节点去维护等等。

从本质上来讲，我们在HTTP接入层要做的就是把数据接进来，做一个producer，导到消息队列里面，中间我们就会做很多高可用、自动恢复、自动扩缩容的事情来保证整个接入层的稳定性。

除了两个底层引擎，还有两个小细节需要说明，一个是CDC(Change Data Capture），一个是Transform Engine。

CDC(Change Data Capture）即变更数据订阅,它是数据库领域非常常见的技术,主要用于订阅数据库的一些变更, 然后可以把变更数据发送到下游，这个技术主要是数据库中才有的概念，数据库包括mysql、pgsql、mongo等。

数据转换引擎（Transform Engine）是指将源数据A转换为源数据B。从技术上来讲，通过自定义代码、logstash、flink这些都可以达到这个功能。Kafka原生内核也集成了这部分功能，同时提供了很多数据处理插件，基于插件实现某个类，这个类就可以实现对应的数据转换功能。

Kafka的转换引擎与其他引擎的区别在于它是非常轻量的，在数据处理里，它的定位是一个简单的数据处理引擎而不是一个全量的数据处理组件。

客户案例

某教育客户需要将直播课学生上下课、签到、浏览等一些行为信息上传到后台进行分析、处理和检索。数据在后台主要有两种业务逻辑：

1. 自定义代码拿到上报数据，进行对应业务逻辑处理；

2. 原始数据进入 Elasticsearch 进行检索分析。

因开发人力有限，希望有一种方便的数据接入服务，简单快速地完成数据的上报、存储。

在经典数据上报架构中，通常需要有如下几个步骤：

1. 搭建/购买存储引擎，用来存储上报的数据；

2. 开发部署数据接收的Server、定义API、运行服务；

3. 定义客户端和服务端的接口协议、鉴权等信息；

4. 客户端根据协议信息编写相应的代码完成数据上报。

在这个4步中，Server端的工作量是最大的，需要考虑代码逻辑开发、server本身扩缩容、接入层本身的稳定性、下游存储的扩缩容和稳定性等等。而当数据量大时，Server端问题会更加明显，需要消耗大量的人力物力来进行维护。

从技术上看，这部分工作通用性很高，DIP可以满足这种场景，提供稳定、弹性、高可靠、高吞吐的数据接入服务。客户可以通过Android、iphone、web这些客户端在我们的SDK把数据直接放到DIP中，DIP目前的缓冲层有两种形态。如果客户用量较大，可以买一个Kafka集群，选自己的topic，不需要为存储付费。量比较小的话也不需要买整个Kafka集群，只需要买单个topic，按量付费。然后进行数据分发，建一个sink，存到es。第二部分会自定义代码，把Kafka协议拿下，这样整个研发成本都可以降下来。

某资讯平台，服务部署在TKE里面，日志通过TKE的采集器投递到了云上的Kafka。这些日志数据主要有三个用途。

部分数据需要进入ElasticSearch进行检索；部分数据经过简单处理后，需要保存到COS进行持久化的存储；部分数据需要进入大数据系统进行业务逻辑方面的处理。当前的做法是写一套大数据处理的Flink逻辑代码，一套处理并转储到ES、COS的代码，然后在两套代码中进行逻辑处理、清洗和分发。

因为Flink 学习维护成本较高，客户早期没有使用Flink时使用的是Logstash ，Logstash在运维过程中，遇到较多性能、稳定性、扩缩容的问题。而且自编码开发迭代投入的人力较多，运维也需要消耗精力。客户希望云上能够有服务能够替换这两部分的工作。如果客户使用DIP数据分发的功能，就可以直接把数据简单处理，直接分发到ES，其余部分只能用Flink，这样就可以省了很多人力成本。

某迅销平台内部有多套系统并行运行，某套系统存储引擎为PGSQL。需要将PGSQL的变更数据存量导入到Elasticsearch里面进行查询。有如下几个需求：

1. 数据写入es的时候需要根据时间分索引；

2. 因为某个数据量大，希望在某个时间区间内只保留某个唯一ID标识的最新数据（update）；

3. 需要根据不同的表将数据分发到不同的索引里面。

自建的架构：PGSQL + Debezium PGSQL + Kafka Connector + Kafka + Logstash + Elasticsearch

DIP架构：      PGSQL + DIP + Elasticsearch

可以看出，客户自建时的链路很长，但如果使用DIP，中间一部分都被替代了，这时候只需要维护两个任务，从研发层面来看，学习成本降低很多。

某保险客户的中台团队迁移上云，因下游团队众多，使用多款MQ产品（Kafka，RocketMQ，RabbitMQ）。各个MQ都是TCP协议接入，有各自的SDK。SDK学习、使用、以及后续切换成本较高。

基于中台考虑，希望上云后能够通过简单的HTTP协议进行接入，屏蔽底层的具体引擎细节。

有三个要求：

1. 简化客户端的使用，最好是HTTP协议；

2. 底层MQ引擎切换对业务无感知；

3. 最好有现成的支持HTTP协议的SDK。

上述客户的需求使用DIP都可以满足。

某内部客户，主要的两部分，业务数大部分都存在MongoDB里面。有一部分客户行为数据，需要上报后进行分析。客户希望将这些数据统一到数据湖(iceberg)进行分析。

自建链路遇到的问题，链路太长，涉及的组件非常多。大多数组件是分布式部署，扩缩容复杂，维护链路的稳定性，透明监控需要花费大量精力。使用DIP后，只需要简单配置，SAAS化，链路的稳定性，扩缩容依托平台处理。

总结

数据接入平台（DIP）主要是为各位开发者介绍数据接入平台是做什么的，在什么场景下可以运用到DIP，底层架构是怎样的。后续的内容会为大家介绍底层各个组件的原理，与其他类似产品的优缺点对比，希望开发者们对腾讯云消息队列CKafka推出的这款产品有更详细的了解，能更好的用到业务中。

《云原生时代的Java应用优化实践》

《全面兼容Eureka：PolarisMesh(北极星)发布1.5.0版本》

扫描下方二维码关注本公众号，

了解更多微服务、消息队列的相关信息！

解锁超多鹅厂周边！