flink实时采集ftp

㈠ 大数据方面核心技术有哪些

大数据技术的体系庞大且复杂，基础的技术包含数据的采集、数据预处理、分布式存储、数据库、数据仓库、机器学习、并行计算、可视化等。

1、数据采集与预处理：

Flume NG实时日志收集系统，支持在日志系统中定制各类数据发送方，用于收集数据；

Zookeeper是一个分布式的，开放源码的分布式应用程序协调服务，提供数据同步服务。

2、数据存储：

Hadoop作为一个开源的框架，专为离线和大规模数据分析而设计，HDFS作为其核心的存储引擎，已被广泛用于数据存储。

HBase，是一个分布式的、面向列的开源数据库，可以认为是hdfs的封装，本质是数据存储、Nosql数据库。

3、数据清洗：MapRece作为Hadoop的查询引擎，用于大规模数据集的并行计算

4、数据查询分析：

Hive的核心工作就是把SQL语句翻译成MR程序，可以将结构化的数据映射为一张数据库表，并提供 HQL(Hive SQL)查询功能。

Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。

5、数据可视化：对接一些BI平台，将分析得到的数据进行可视化，用于指导决策服务。

㈡ 流批一体不只有Flink，还有实时数据模型

通常来讲，数据仓库的建设，都是以离线作为主要的密报，下游的应用，不论是报表还是接口，所提供的数据也大多是T-1时效性。

但伴随着业务的变化，当离线做到没什么可以继续做的时候，实时就会被拿出来，作为新一个阶段的目标进行攻克。

在流批一体建设之前，这种实时诉求通常会开发成分钟级的任务，通过近实时的方案来解决业务的问题，但分钟级会带来诸如任务过多、资源挤占较大、无法支持复杂逻辑等问题。

因此专门支持实时计算的框架，比如早期的Storm，能够尝试从纯实时的角度解决业务问题，就被拿出来作为尝试。然而Storm的局限性也很大，因为那会的任务开发只能通过Java的方式来进行，与Hive所推崇的纯SQL方案相比，上手难度大了不少，同时两套代码的逻辑几乎没有可比性，这种方案也就一直没有什么声音。

尽管实时技术有各种缺陷，但作为一种能够很容易讲清楚价值的项目，同时又非常便于向上汇报的技术方案，实时技术还是被或多或少的做了起来。在大多数的公司里，实时和离线就会有不同的团队进行维护，或者是同一个团队，但分成不同的项目来执行。这个阶段，优先高效的把业务做起来，哪怕场景再简单，但能够证明实时有价值和前景，这个阶段的目标就算完成了。

以上的各种方案，难免会带来三个特别难以解决的问题：

（1）数据的口径上，实时和离线很容易不统一；
（2）数据模型的规范上，实时和离线也往往是分开建设；
（3）即便是同一种口径和同一种规范，实时和离线也要分成两套代码来维护。

这三个问题短时间内会被高速发展掩盖掉，但当业务对实时的诉求越来越多、压力越来越大的时候，口径和代码的不统一，就会越来越成为阻碍敏捷开发的障碍，需要有方案进行解决。

后来Flink出现了，带来了流批一体的全新方案，这个问题便出现了解决的曙光，这也比较接近我们对于实时计算的理想方案，因为其意义堪比Hive，也成为了各个大厂面试的标配问题。

然而，仅仅学会Flink是不够的，因为流批一体带来的并不仅仅是技术方案或者是框架的改变，同样带来了数据模型的改变，这就要求我们从数据模型上，而不是技术方案上，来制定我们的实时方案。

那么我们如何理解“实时数据模型”这件事情呢？

通常而言，我们关心的内容，包括如下几个方面：

（1）实时数据源与离线数据源存在差异，导致相同的字段，取值或者类型会存在不相等的情况；
（2）实时和离线由于底层执行机制的不同，通常需要维护两套代码，会带来诸如口径不统一、质量检测难的问题；
（3）产品逻辑变化较快时，离线模型修改相对容易，但实时模型需要考虑压测、削峰、重启等技术问题，维护成本非常高昂。

数据仓库之所以能够普及并被业务接受，正是因为其模型能够屏蔽掉底层差异的问题，并且有相对可靠的数据质量监控方法，并且变更成本非常低。而实时数仓如果想要替代掉离线数仓，以上的问题通常是需要一些模型设计甚至是平台工具的来解决，这些问题解决的重要性，并不比Flink弱。

我们先从比较可控的模型层面说起。

在离线的概念里，数仓模型设计成了DWD/DWS/ADS三个层级，原本的概念是DWD面向事实表的构建，DWS面向公共指标的统一，ADS负责灵活的口径变化问题。

在离线的概念里，DWD/DWS/ADS三个层级需要保留，但负责的目标会有一些变化，同时还需要增加存储统一层，也就是以TiDB/Holo为代表的数据库，来承担服务分析一体化的诉求。

让我们先看DWD层，DWD承担了屏蔽实时离线链路差异的问题，最重要的作用是保证表结构的统一及字段内容的对齐。DWD最重要的意义，是保证离线表和实时表，其表结构和字段概念是相同的。

为什么这么强调？试想一下，在离线场景下，我们可以在DWD上灵活的增加各种统计标签，或者是将维度退化到事实表，都是一些left join或者是服务端直接打标可以解决的事情。但在实时场景下，这会变成多流join或者是缓存等更复杂的技术场景，导致这些信息并不能有效的记录到DWD，因此DWD的设计就要产生一些变化，有一些内容在实时场景下无法准确记录，这一类信息需要标识到对应的字段描述上，下游使用时才不会出错。

同时，实时和离线存储数据的介质，也必然有一些区别。例如离线可以存在HDFS上，实时则可能视情况保存在数据库、HDFS甚至是内存中，这时候对于字段格式、读取方式都会有差异，设计表时其约束条件也会更多。

因而，DWD更多承担了逻辑统一的职责，依旧以事实表为基础，但约束条款要比离线更多。

再看一下DWS层，离线上DWS是负责口径统一的重要一环，将通用的维度和口径计算方法抽象出现，以提供跨数据域的灵活使用。但在实时场景下，这一类的维护收益通常都比较低，不仅因为实时只看当天的数据，也是因为实时本身的维度难度就较大，多一层模型其收益会急速下降，因而大多数时候会忽略掉DWS的建设，ADS直接引用DWD进行统计。

然而，DWS毕竟存储的内容要比DWD少很多，因此如果计算资源瓶颈非常明显，或者是业务场景不需要分析实时明细数据的情况下，或者是DWD的下游引用过多时，DWS可以承担削峰的重任，通过减少数据量以应对大促等场景，还是有一定意义的。

接下来就是最重要的ADS层，在这一层上，逻辑统一、口径统一、大促削峰在前置模型上都得到了一定程度的解决，ADS则像离线一样承担了应对需求变化的重任。

但ADS所面临的情况和离线还是有所不同的，因为ADS的任务启动，不仅要启动一个离线的跑批任务，还要同时启动一个实时的流式任务，而ADS往往会同时统计离线+实时的结果，以应对同比、环比等场景。

这时候很多具体Case要具体分析了，因为特定场景的坑会非常多。例如最常见的“同比”，要对比今年和去年的结果变化，离线往往会统计分小时的结果，但实时会累计起始时刻到当期时刻的结果，因而当一个小时没有结束的时候，这个同比的波动变化会非常大，给人一种“数据是错误的”印象，新手很容易踩这个坑，从而被业务质疑。

因此，针对累计统计指标，从代码设计上就要考虑到这种情况，都根据时间字段统计起始到当前时刻的结果的，在代码逻辑上会要求一些统计技巧。

很多时候，因为业务指标变化太快，改实时代码是来不及的，这时候一部分的工作量甚至需要报表工具的数据集来解决，改动查询sql，要比改动任务来的快捷多了。但这部分的能力，其实是依赖于存储工具的，个人认为可以分到存储统一层来解决。

最后是存储统一层，因为一些特殊的场景，比如实时分析明细数据，或者是不确定时间周期的多天统计结果，如果依赖Flink SQL来解决是有些不现实的，因而这部分的压力需要数据库来承担。

简单讲，就是将明细做轻度的汇总后，直接写到数据库，实时更新，下游自定义条件，并直接读库统计结果。这种场景既要求数据库有OLAP的计算能力，也要有OLTP的稳定特点，因而TiDB和Holo这一类HTAP的引擎就变得非常重要。

因为多了实时的部分，因此过去面向离线的开发工具，也需要有一些特定的改造，以适应实时的开发和运维诉求。

对于开发工具而言，其目标集中在四个场景上：元数据定义与获取、数据建模、开发与测试、运维与监控。

其次讲数据建模，因为建模的理论已经稳定了有些年头了，绝大多数场景下都是按照既定的方案来执行。过去离线当道时，规范执行的弱一些不是什么大问题，但流批一体当道的年代，规范是需要强约束的，这就对了开发工具提出了一定的要求，是否能够从平台层面上对规范进行内置，并以此来约束开发的同学，降低不规范模型对后期维护带来的压力。

这种建模能力的代表有两种，一种是规范表的命名，填写相应的分层+主题域+数据域+统计刷新方式，从源头上规范表的目标和作用；一种是规范指标的定义和使用，例如原子指标还是派生指标，统计周期多少，业务限定用语如何规范，统计粒度怎么填写。

在实际开发中，通过工具的限制，如果规范可以做的好，代码是可以自动生成出来的。当然，以上的功能，都属于通过牺牲开发效率，来提升数据质量的范畴，使用时需要根据团队的情况来限定。

再次是开发和测试，这是平台提供的最重要的能力。在开发层面，就是代码的预编译能力+发布功能。预编译不仅要检查代码的逻辑是否正确，同时对于代码中依赖的其他数据源，获取到的元数据信息是否准确，至少字段的命名不会有大的问题。当代码预编译通过，发布上线后，还需要检测当前是否有资源支持任务启动，并且上游的消息队列是否是启动的状态。

实时的测试一直都是比较大的问题，它不像离线可以启动一个SQL任务看看结果，实时在每个阶段的输入和输出，是需要通过平台支持的日志打印功能来进行辅助的。很多时候我们会新建一个测试专用的topic来测试结果，但对于流量较大的线上任务而言，这种方式无法像离线区分Dev环境一样，能够对资源进行隔离，因而如果能够支持圈定数据的输入和打印输出，对于测试的效率而言无疑是最佳的。

最后要提到的是运维与监控能力。运维能力是指根据输入的RPS，或者是cu使用情况，或者是任务的整体延迟，提供相应的参数调优能力，通过参数来调整任务的执行情况。并且能够根据以上指标的变化，自定义相应的阈值，提供相应的告警能力，通过短信或者是消息工具的方式触达任务维护者。

实时与离线有一些不同的是，离线可以通过增加一个监控节点的方式，通过group by判断数据是否重复，而实时任务则非常依赖Flink自身的一致性能力，因而发现和解决问题的成本更高。

其实做到运维这个环节，对人的要求其实是更高的。因为流批一体在运维上会带来一个好处，即实时任务和离线任务能够错峰执行，实时在白天压力大，而离线在晚上压力大。但同样的，这种方式对于维护者而言更加痛苦，因为不仅晚上要熬夜值班，白天同样不能休息，在大促期间甚至需要轮班来维护任务，可以说是“汇报一时爽，痛苦长相伴”。

从远处来看，流任务和批任务，在自身的机制上就存在非常大的差异，批程序面上的是特定时间内相对静态的数据，而流程序处理的则是change-log，虽然有可能数据在表结构层面，通过数据模型的设计来保持一致，但是在语义层面，其根本还是不一样的。这一点可能是最制约批流一体发展的问题，也是最难实现统一或者永远也不可能统一的。

综上，对于实时模型，开发工具需要将监控实时部分的能力进行补全，就像DWD层需要分别维护实时和离线两套架构一样，开发工具也需要分别维护两套架构的结果，因而现阶段的实时开发，还做不到降低维护和开发的成本，只能减轻其中部分环节的工作量。

以上讲了很长时间的实时模型，但从实际的效果上看，业务并不会感知到多么明显的技术变化，相反会有一种“面子工程”的感觉在里面。

当然，我并不否认实时的价值，在“搜广推”这三个技术占主导的领域内，作用还是很大的。但实时毕竟要比离线的内容，更加的难以理解，出现问题的排查成本也更高。这种复杂性使得我们在应对变化时，往往做不出有效的应对，就会变得特别被动。

因而，说一句事后的话，就是“实时的价值取决于业务方，而不是技术方”。只有业务对实时痛点强烈的场景下，我们做如此复杂的研究和应对，才能体现出自己的价值，更多的时候，是在“王婆卖瓜，自卖自夸”。有这种投入，还不如多招几个分析师更靠谱和实在。

本人之前的文章《天下数据，唯快不破》，重点强调了一个“快”字。但“天下熙熙皆为利来，天下攘攘皆为利往”，这个快更多的是在讲应对“变化”的快，而不是“技术”自己的快。

所以，为了以后的职业发展，我们要跟进实时技术的变化，但从自身的工作角度出发，如何应对业务的变化，才是自己要关心的课题。

㈢ 基于flink sql构建实时数据仓库

根据目前大数据这一块的发展，已经不局限于离线的分析，挖掘数据潜在的价值，数据的时效性最近几年变得刚需，实时处理的框架有storm，spark-streaming，flink等。想要做到实时数据这个方案可行，需要考虑以下几点：1、状态机制 2、精确一次语义 3、高吞吐量 4、可弹性伸缩的应用 5、容错机制，刚好这几点，flink都完美的实现了，并且支持flink sql高级API，减少了开发成本，可用实现快速迭代，易维护等优点。

离线数仓的架构图：

实时数仓架构图:

目前是将实时维度表和DM层数据存于hbase当中，实时公共层都存于kafka当中，并且以写滚动日志的方式写入HDFS（主要是用于校验数据）。其实在这里可以做的工作还有很多，kafka集群，flink集群，hbase集群相互独立，这对整个实时数据仓库的稳定性带来一定的挑战。

一个数据仓库想要成体系，成资产，离不开数据域的划分。所以参考着离线的数据仓库，想着在实时数仓做出这方面的探索，理论上来讲，离线可以实现的，实时也是可以实现的。 并且目前已经取得了成效，目前划分的数据域跟离线大致相同，有流量域，交易域，营销域等等。当然这里面涉及到维表，多事务事实表，累计快照表，周期性快照表的设计，开发，到落地这里就不详述了。

维度表也是整个实时数据仓库不可或缺的部分。从目前整个实时数仓的建设来看，维度表有着数据量大，但是变更少的特点，我们试想过构建全平台的实时商品维度表或者是实时会员维度表，但是这类维度表太过于复杂，所以针对这类维度表下面介绍。还有另外一种就是较为简单的维度表，这类维度可能对应着业务系统单个mysql表，或者只需要几个表进行简单ETL就可以产出的表，这类维表是可以做成实时的。以下有几个实施的关键点：

如下是离线数据同步架构图：

实时数据的接入其实在底层架构是一样的，就是从kafka那边开始不一样，实时用flink的UDTF进行解析，而离线是定时（目前是小时级）用camus拉到HDFS，然后定时load HDFS的数据到hive表里面去，这样来实现离线数据的接入。实时数据的接入是用flink解析kafka的数据，然后在次写入kafka当中去。

由于目前离线数据已经稳定运行了很久，所以实时接入数据的校验可以对比离线数据，但是离线数据是小时级的hive数据，实时数据存于kafka当中，直接比较不了，所以做了相关处理，将kafka的数据使用flink写HDFS滚动日志的形式写入HDFS，然后建立hive表小时级定时去load HDFS中的文件，以此来获取实时数据。

完成以上两点，剩余还需要考虑一点，都是小时级的任务，这个时间卡点使用什么字段呢?首先要确定一点就是离线和实时任务卡点的时间字段必须是一致的，不然肯定会出问题。目前离线使用camus从kafka将数据拉到HDFS上，小时级任务，使用nginx_ts这个时间字段来卡点，这个字段是上报到nginx服务器上记录的时间点。而实时的数据接入是使用flink消费kafka的数据，在以滚动日志的形式写入HDFS的，然后在建立hive表load HDFS文件获取数据，虽然这个hive也是天/小时二级分区，但是离线的表是根据nginx_ts来卡点分区，但是实时的hive表是根据任务启动去load文件的时间点去区分的分区，这是有区别的，直接筛选分区和离线的数据进行对比，会存在部分差异，应当的做法是筛选范围分区，然后在筛选nginx_ts的区间，这样在跟离线做对比才是合理的。

目前实时数据接入层的主要时延是在UDTF函数解析上，实时的UDTF函数是根据上报的日志格式进行开发的，可以完成日志的解析功能。

解析流程图如下：

解析速率图如下：

该图还不是在峰值数据量的时候截的，目前以800记录/second为准，大概一个记录的解析速率为1.25ms。
目前该任务的flink资源配置核心数为1，假设解析速率为1.25ms一条记录，那么峰值只能处理800条/second，如果数据接入速率超过该值就需要增加核心数，保证解析速率。

介绍一下目前离线维度表的情况，就拿商品维度表来说，全线记录数将近一个亿，计算逻辑来自40-50个ods层的数据表，计算逻辑相当复杂，如果实时维度表也参考离线维度表来完成的话，那么开发成本和维护成本非常大，对于技术来讲也是很大的一个挑战，并且目前也没有需求要求维度属性百分百准确。所以目前（伪实时维度表）准备在当天24点产出，当天的维度表给第二天实时公共层使用，即T-1的模式。伪实时维度表的计算逻辑参考离线维度表，但是为了保障在24点之前产出，需要简化一下离线计算逻辑，并且去除一些不常用的字段，保障伪实时维度表可以较快产出。

实时维度表的计算流程图：

目前使用flink作为公司主流的实时计算引擎，使用内存作为状态后端，并且固定30s的间隔做checkpoint，使用HDFS作为checkpoint的存储组件。并且checkpoint也是作为任务restart以后恢复状态的重要依据。熟悉flink的人应该晓得，使用内存作为状态后端，这个内存是JVM的堆内存，毕竟是有限的东西，使用不得当，OOM是常有的事情，下面就介绍一下针对有限的内存，如果完成常规的计算。

㈣ Flink 原理详解

Flink 是一个流处理框架，支持流处理和批处理，特点是流处理有限，可容错，可扩展，高吞吐，低延迟。

流处理是处理一条，立马下一个节点会从缓存中取出，在下一个节点进行计算

批处理是只有处理一批完成后，才会经过网络传输到下一个节点

流处理的优点是低延迟 批处理的优点是高吞吐

flink同时支持两种，flink的网络传输是设计固定的缓存块为单位，用户可以设置缓存块的超时值来决定换存块什么时候进行传输。 数据大于0 进行处理就是流式处理。
如果设置为无限大就是批处理模型。

Flink 集群包括 JobManager 和 TaskManager .

JobManager 主要负责调度 Job 并协调 Task 做 checkpoint，职责上很像 Storm 的 Nimbus。从 Client 处接收到 Job 和 JAR 包 等资源后，会生成优化后的执行计划，并以 Task 的单元调度到各个 TaskManager 去执行。

TaskManager 在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobManager 处接收需要 部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。

flink on yarn 是由client 提交 app到 RM 上， 然后RM 分配一个 AppMaster负责运行 Flink JobManager 和 Yarn AppMaster, 然后 AppMaster 分配 容器去运行 Flink TaskManger

SparkStreaming 是将流处理分成微批处理的作业， 最后的处理引擎是spark job

Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块，Spark Streaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据。每个块都会生成一个Spark Job处理，然后分批次提交job到集群中去运行，运行每个 job的过程和真正的spark 任务没有任何区别。

JobScheler, 负责 Job的调度通过定时器每隔一段时间根据Dstream的依赖关系生一个一个DAG图

ReceiverTracker负责数据的接收，管理和分配
ReceiverTracker在启动Receiver的时候他有ReceiverSupervisor,其实现是ReceiverSupervisorImpl, ReceiverSupervisor本身启 动的时候会启动Receiver，Receiver不断的接收数据，通过BlockGenerator将数据转换成Block。定时器会不断的把Block数据通会不断的把Block数据通过BlockManager或者WAL进行存储，数据存储之后ReceiverSupervisorlmpl会把存储后的数据的元数据Metadate汇报给ReceiverTracker，其实是汇报给ReceiverTracker中的RPC实体ReceiverTrackerEndpoin

spark on yarn 的cluster模式， Spark client 向RM提交job请求, RM会分配一个 AppMaster, driver 和 运行在AppMAster节点里， AM然后把Receiver作为一个Task提交给Spark Executor 节点， Receive启动接受数据，生成数据块，并通知Spark Appmaster, AM会根据数据块生成相应的Job, 并把Job 提交给空闲的 Executor 去执行。

1：需要关注流数据是否需要进行状态管理
2：At-least-once或者Exectly-once消息投递模式是否有特殊要求
3：对于小型独立的项目，并且需要低延迟的场景，建议使用storm
4：如果你的项目已经使用了spark，并且秒级别的实时处理可以满足需求的话，建议使用sparkStreaming
5：要求消息投递语义为 Exactly Once 的场景；数据量较大，要求高吞吐低延迟的场景；需要进行状态管理或窗口统计的场景，建议使用flink

Flink 提供的Api右 DataStream 和 DataSet ，他们都是不可变的数据集合，不可以增加删除中的元素， 通过 Source 创建 DataStream 和 DataSet

在创建运行时有：

Flink的每一个Operator称为一个任务， Operator 的每一个实例称为子任务，每一个任务在JVM线程中执行。可以将多个子任务链接成一个任务，减少上下文切换的开销，降低延迟。

source 和 算子map 如果是 one by one 的关系，他们的数据交换可以通过缓存而不是网络通信

TaskManager 为控制执行任务的数量，将计算资源划分多个slot,每个slot独享计算资源，这种静态分配利于任务资源隔离。

同一个任务可以共享一个slot, 不同作业不可以。

这里因为 Source 和 Map 并行度都是4 采用直连方式，他们的数据通信采用缓存形式

所以一共需要两个TaskManager source,Map 一个，rece一个， 每个TaskManager 要3个slot

JobManager 将 JobGraph 部署 ExecutionGraph

设置的并行度，可以让一个ExecJobVertex 对应 多个并行的ExecVertex 实例。

Flink通过状态机管理 ExecGraph的作业执行进度。

Flink 将对象序列化为固定数量的预先分配的内存段，而不是直接把对象放在堆内存上。
Flink TaskManager 是由几个内部组件组成的：actor 系统（负责与 Flink master 协调）、IOManager（负责将数据溢出到磁盘并将其读取回来）、MemoryManager（负责协调内存使用。

数据源：

Sink:

时间：

处理时间：取自Operator的机器系统时间

事件时间： 由数据源产生

进入时间： 被Source节点观察时的系统时间

如果数据源没有自己正确创建水印，程序必须自己生成水印来确保基于事件的时间窗口可以正常工作。。

DataStream 提供了 周期性水印，间歇式水印，和递增式水印