All wisdom begins with memory.

之前已经聊过了目前市面上常用的一些架构及技术选型。 传统数据入仓 - 离线方向 MySQL→Sqoop→HDFS→Hive 传统数据入仓架构 1.0，主要使用 DataX 或 Sqoop 全量同步到 HDFS，再围绕 Hive 做数仓。 此方案存在诸多缺陷：容易影响业务稳定性，因为每天都需要从业务表里查询数据；天级别的产出导致时效性差，延迟高；如果将调度间隔调成几分钟一次，则会对源库造成非常大的压力；扩展性差，业务规模扩大后极易出现性能瓶颈。 传统数仓2.0 - 增加实时方向(Canal、dataX实时采集增量数据到Kafka上再Sink到HDFS上，最后增量全量做合并，最终还是围绕Hive) 分为实时和离线两条链路，实时链路做增量同步，比如通过 Canal 同步到 Kafka 后再做实时回流； 全量同步一般只做一次，与每天的增量在 HDFS 上做定时合并，最后导入到 Hive 数仓里。 此方式只做一次全量同步，因此基本不影响业务稳定性，但是增量同步有定时回流，一般只能保持在小时和天级别，因此它的时效性也比较低。同时，全量与增量两条链路是割裂的，意味着链路多，需要维护的组件 ...

传统方案传统数据集成方案的痛点 大数据技术的应用可以从海量的用户行为数据中进行挖掘分析，根据分析结果优化平台的服务质量，最终满足用户的需求。大数据分析平台就是将大数据技术应用于教育培训领域，为企业经营提供数据支撑： 建立集团数据仓库，统一集团数据中心，把分散的业务数据进行预先处理和存储。 根据业务分析需要，从海量的用户行为数据中进行挖掘分析，定制多维的数据集合，形成数据集市，供各个场景主题使用。 前端业务数据展示选择和控制，选取合适的前端数据统计、分析结果展示工具。 - 上图为传统数据入仓架构 1.0，主要使用 DataX 或 Sqoop 全量同步到 HDFS，再围绕 Hive 做数仓。 此方案存在诸多缺陷：容易影响业务稳定性，因为每天都需要从业务表里查询数据；天级别的产出导致时效性差，延迟高；如果将调度间隔调成几分钟一次，则会对源库造成非常大的压力；扩展性差，业务规模扩大后极易出现性能瓶颈。 上图为传统数据入仓 2.0 架构。分为实时和离线两条链路，实时链路做增量同步，比如通过 Canal 同步到 Kafka 后再做实时回流；全量同步一般只做一次，与每天的增量在 HDFS ...

https://www.mubu.com/doc/Ud_30MJRFa

Flink 中的状态编程 在 Flink 中，算子任务可以分为无状态和有状态两种情况。 在传统的事务型处理架构中，这种额外的状态数据是保存在数据库中的。而对于实时流处理来说，这样做需要频繁读写外部数据库，如果数据规模非常大肯定就达不到性能要求了。所以 Flink 的解决方案是，将状态直接保存在内存中来保证性能，并通过分布式扩展来提高吞吐量。 有状态算子的一般处理流程： 算子任务接收到上游发来的数据； 获取当前状态； 根据业务逻辑进行计算，更新状态； 得到计算结果，输出发送到下游任务。 状态分类按照由 Flink 管理还是用户自行管理，状态可以分为原始状态 ( Raw State ) 和托管状态 (Managed State)。 原始状态：即用户自定义的 State。Flink 在做快照的时候，把整个 State 当做一个整体，需要开发者自己管理，使用 byte 数组来读写状态内容。 托管状态：是由 Flink 框架管理的 State，如 ValueState、ListState 等，其序列化和反序列化由 Flink 框架提供支持，无需用户感知、干预。通常在 DataStre ...

Flink 中的水印操作问题引入流处理中的乱序问题当 flink 以 EventTime 模式处理流数据时，它会根据数据里的时间戳来处理基于时间的算子。 但是由于网络、分布式等原因，会导致数据乱序的情况。 watermark解决乱序问题不以事件时间作为触发计算的条件，而是根据Watermark判断是否触发。 当Watermark的时间戳等于Event中携带的EventTime时候，上面场景（Watermark=EventTime)的计算结果如下： Watermark=EventTime Watermark = EventTime -5s如果想正确处理迟来的数据可以定义Watermark生成策略为 Watermark = EventTime -5s， 如下： 基于SQL的水印实现场景： 使用Socket模拟接收数据 设置WaterMark，设置的逻辑：在第一条数据进来时，设置WaterMark为0，指定第一条数据的时间戳后，获取该时间戳与当前 WaterMark的最大值，并将最大值设置为下一条数据的WaterMark，以此类推 12345 ...

准备工作12345show databases ;create database if not exists db;use db;-- 一些语句会走 MapReduce，所以慢。 可以开启本地化执行的优化。set hive.exec.mode.local.auto=true;-- (默认为false) 访问量统计准备数据+需求分析12345678910111213141516171819202122232425CREATE TABLE db.test1 ( userId string, visitDate string, visitCount INT ) ROW format delimited FIELDS TERMINATED BY "\t";INSERT overwrite TABLE db.test1VALUES ( 'u01', '2017/1/21', 5 ), ( 'u02', '2017/1/23', 6 ), ( 'u03 ...

数据治理思考数据问题: 数据管理责任不清晰,造成数据问题无人决策解决; 数据多源头,造成数据不一致,不可信; 数据大量搬家造成IT重复投资; 数据无定义造成难于理解、难于使用; 各部门发布报告,统计口径不一致,困扰业务决策; 数据形态多样化,数据量迅猛增长,数据处理逻辑复杂,投资大; 华为在数字化转型过程中,解决了上述问题,因为华为认识到只有建立了完整的数据治理体系,保证数据内容的质量,才能够真正有效地挖掘企业内部的数据价值,对外提高竞争力。 数据治理模块域数据治理主要专注于如下模块域: 数据集成 数据集成用来完成数据入湖动作,不是简单的数据搬家,而是按照一定的方法论 进行数据备份。数据入湖的前提条件是满足6项数据标准,包括:明确数据 Owner、发布数据标准、定义数据密级、明确数据源、数据质量评估、元数据注 册。此标准由数据代表在入湖前完成梳理并在数据治理平台上进行资产注册。 数据标准 数据标准管理着重建立统一的数据语言,L1到L5数据层级业务对象的定义是数据 标准的载体,并对应发布包括L1到L5数据层级的数据标准。各业务对象对应物理 实现的IT系统 ...

Flink SQL基本介绍实操FlinkSQL启动!!1234# 先启动集群/export/server/flink-1.15.2/bin/start-cluster.sh# 再启动客户端/export/server/flink-1.15.2/bin/sql-client.sh 设置输出模式设置输出模式：我一般用的是tableau模式. 123456# 表格模式（table mode）在内存中实体化结果，并将结果用规则的分页表格可视化展示出来。执行如下命令启用：SET sql-client.execution.result-mode=table;# 变更日志模式（changelog mode）不会实体化和可视化结果，而是由插入（+）和撤销（-）组成的持续查询产生结果流：SET sql-client.execution.result-mode=changelog;# Tableau模式（tableau mode）更接近传统的数据库，会将执行的结果以制表的形式直接打在屏幕之上：SET sql-client.execution.result-mode=tableau; 全心全意写SQ ...

粥所周知, Flink程序由四部分构成, 运行环境 + Source + Transformation + Sink. 接下来一个个说. 运行环境批处理 获取批处理执行环境（用于测试/生产） 1ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); 流处理 获取流式处理执行环境（用于测试/生产） 1StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 流批一体 获取流批一体处理执行环境（用于测试/生产） 123final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置为批处理模式env.setRuntimeMode(ExecutionMode.BATCH); 本地环境 创建本地执行环境（用于 ...

作业(Job)提交流程高层级抽象视角 Flink 的提交流程，随着部署模式、资源管理平台的不同，会有不同的变化。首先我们从一个高层级的视角，来做一下抽象提炼，看一看作业提交时宏观上各组件是怎样交互协作的。 具体步骤如下： （1） 一般情况下，由客户端（App）通过分发器提供的 REST 接口，将作业提交给JobManager。 （2）由分发器启动 JobMaster，并将作业（包含 JobGraph）提交给 JobMaster。 （3）JobMaster 将 JobGraph 解析为可执行的 ExecutionGraph，得到所需的资源数量，然后向资源管理器请求资源（slots）。 （4）资源管理器判断当前是否由足够的可用资源；如果没有，启动新的 TaskManager。 （5）TaskManager 启动之后，向 ResourceManager 注册自己的可用任务槽（slots）。 （6）资源管理器通知 TaskManager 为新的作业提供 slots。 （7）TaskManager 连接到对应的 JobMaster，提供 slots。 （8）JobMaster 将需要执行的 ...