摘要：DAP数据分析平台是面向企业数据治理和数据分析的大数据应用平台，通过企业数仓建设和数据可视化分析，实现企业数据的整合归集、标准统一，构建数据资产，支撑企业业务管理和决策。在强化数据整合、管控的同时，实现数据的集成联动，通过流动的数据服务于业务，产生业务价值，实

DAP数据分析平台是面向企业数据治理和数据分析的大数据应用平台，通过企业数仓建设和数据可视化分析，实现企业数据的整合归集、标准统一，构建数据资产，支撑企业业务管理和决策。在强化数据整合、管控的同时，实现数据的集成联动，通过流动的数据服务于业务，产生业务价值，实现数据价值的最大化。

在数据集成整合的过程中，DAP平台需要从各个内外部系统数据源中采集各类数据，包括结构化数据、半结构化数据和非结构化数据等，同时为了满足不同业务和使用的需要，在采集方式上需要实现实时采集和离线采集相结合的方式。在上篇中主要介绍了Kafka部署过程，本篇将主要介绍Flink部署及整体测试过程。

本次主要介绍Flink虚拟机部署的方式，如果整体环境采用K8S容器化部署，建议直接部署Flink容器，可以通过UMC实现更加方便的部署和管理。

1.部署过程

1.下载安装包，上传；

2.解压：tar -zxvf flink-1.14.6-bin-scala_2.11.tgz；

3.进入flink目录，启动：./bin/start-cluster.sh；

4.端口检查；

5.访问测试。

2.测试验证

1.读写测试：

2.执行结果：

3.控制台显示：

3.ESB测试

1.修改全局变量：isLocal改为“Y”，采用本地模式（非k8s容器部署）；

2.修改访问地址：flinkRestURLPrefix改为Flink访问地址（http://ip:port），端口默认为8081（如果修改了配置文件，以修改后为准）；

3.通过Flink的Web UI上传jar文件；

4.通过接口获取JarID：get方法调用http://:/jars，返回jar列表，获取JarID；

5.修改全局变量；

6.配置任务；

7.提交测试。

根据实时集成的测试流程，基于Flink和Kafka进行整体测试，模拟源头系统的数据变化，通过Flink采集源头的数据变更，并将变更写入Kafka，再通过ESB消费Kafka数据写入DAP的ODS表中。

1.测试方案

1.配置数据源，包括biz、ods和edw，其中biz和ods用mysql，edw用doris；

2.在DAP的biz库中建立业务来源表；

3.在DAP中通过biz配置ODS表，同步方式选择“外部同步”；

4.在SMC中Flink实施采集，通过以biz中表为源头，通过cdc的方式将数据采集到Kafka中；

5.通过ESB设计器手动创建Queue流程，读取Kafka中的数据，并进行数据格式转换，再写入dap的ODS表中；

6.在事实配置中创建事实表，目标数仓为Doris数仓，通过ODS表字段构建数仓基础事实表；

7.在ESB设计器中通过流程向导创建“ODS转换到EDW”的Http流程，并进行部署；

8.调整Kafka到ODS的Queue流程，在写入ODS表后添加HttpInvoker节点，调用“ODS转换到EDW”的加工流程，实现ODS到EDW的加工；

9.在biz表中添加、修改数据，验证ODS以及Doris数仓中的数据变化；

10.加工后检查数仓加工日志；

11.检查ESB后台日志的运行情况。

2.配置过程

1.配置DAP数据源；

2.创建biz表；

3.创建ODS表；

4.配置Flink采集；

5.创建消费的Queue流程；

6.配置事实表；

7.创建加工流程；

8.通过流程调用加工流程；

9.添加数据；

10.修改数据。

3.注意事项

1.Flink日志检查；

1)客户端日志：flink/log/flink-appuser-client-主机名.log；

2)服务端日志：flink-appuser-standalonesession-2-主机名.log；

3)Sql-client（bin/sql-client.sh）日志：flink-appuser-sql-client-主机名.log；

4)Job日志：启动job时指定的日志文件，该文件会输出job中打印的输出信息。

2.Kafka日志：

1)Broker日志：kafka/logs/server.log。

3.ODS同步方式采用“外部同步”时，需要手动创建流程，并且创建Queue流程进行数据消费。

基于ESB数据总线进行数据采集是DAP平台实现数据采集、加工、转换、汇总的基础，而根据实际业务、需求以及数据类型的不同，采集方式也会有所区别，本次数据针对结构化数据的实时采集进行介绍，通过Flink与Kafka结合的方式实现的采集过程。

1.问题总结

由于Flink、Kafka组件都是基于虚拟机部署的，并且都是最新版本，在实际应用时需要结合使用，同时采集过程涉及到了CDC的实时数据读取，所以在整体测试过程中需要根据运行要求补充相关的依赖jar，从而保证运行的准确性。

由于Flink是直接通过ESB的配置进行调用的，所有在ESB的Flink任务运行前，需要将ESB专用FlinkJob文件放到Flink的环境中并进行启用，从而保证ESB的任务可以顺利执行。

2.产品功能

在产品层面，ESB和DAP相互结合已经完成了对Flink实时采集的支持，通过ESB配置Flink任务采集来源数据，再基于流程进行数据处理，写入DAP的数仓中，从而实现数据整体的采集过程。除了本次介绍的Fink结合Kafka的采集模式，在ESB中还预置了数据库到数据库、文件到数据库等不同的集成方式，可以满足不同业务场景的需求。

3.说在最后

DAP数据分析平台作为数据治理和数据分析平台，提供完整的数据采集、加工、转换、汇总功能，从而实现数仓的构建。同时DAP提供了数据可视化、数据服务、数据预测、数据资产等功能，可以深度对数仓数据进行应用，从而满足各类业务场景的数据使用需求。

在实际使用过程中，DAP数据分析平台和ESB数据总线平台结合使用，通过ESB快速生成采集流程，实现数据采集和数仓构建。为了保证数仓采集数据的准确性，以及基础数据的一致性，会在基础方案的基础上添加MDM平台构建一体化数据中台，从而实现集主数据治理、数仓建设、数据分析为一体的数据中台方案，共同完成数据治理的全过程，从而为企业的数据体系建设提供全面的支持。

本文由@数通畅联原创，欢迎转发，仅供学习交流使用，引用请注明出处！谢谢~

来源：数通畅联