数据处理

数据集成

首先要有可靠、完整的数据流，后续才能谈得上在上层对这些数据进行存储、可视化、分析

ELT相比ETL，因为转换是在目的地，所以可以根据后续使用的情况做调整，比较灵活

数据清洗

需要一些清洗规则帮助确保数据质量和一致性，减少数据中的错误、不完整、重复等问题，使得数据标准、干净、连续

数据变换

  1. 数据平滑：去除数据中的噪声，将连续数据离散化
  2. 数据聚集：对数据进行汇总
  3. 数据概化：将数据由较低的概念抽象成为较高的概念，减少数据复杂度，即用更高的概念替代更低的概念。比如说上海、杭州、深圳、北京可以概化为中国
  4. 数据规范化：使属性数据按比例缩放，这样就将原来的数值映射到一个新的特定区域中
    1. Min-max 规范化：将原始数据变换到[0,1]的空间中
    2. Z-Score 规范化：新数值 =（原数值 - 均值）/ 标准差
    3. 小数定标规范化：移动小数点的位置来进行规范化。小数点移动多少位取决于属性 A 的取值中的最大绝对值
  5. 属性构造：构造出新的属性并添加到属性集中

批处理

UNIX的管道

MapReduce的不足

需要的：

  1. 一种技术抽象让多步骤数据处理变得易于维护
  2. 不要复杂的配置，需要能自动进行性能优化
  3. 要能把数据处理的描述语言，与背后的运行引擎解耦合开来
  4. 要统一批处理和流处理的编程模型
  5. 要在架构层面提供异常处理和数据监控的能力

20221214142627

分布式批处理需要解决的问题

  1. 如何将输入数据分区
  2. 容错：任务可能随时会失败

流处理

消息系统

早期使用消息队列来实现流处理：

stateDiagram
  direction LR
  数据源 --> 消息队列1
  消息队列1 --> 处理逻辑1
  处理逻辑1 --> 消息队列2
  消息队列2 --> 处理逻辑3
  消息队列2 --> 处理逻辑4
  处理逻辑3 --> 消息队列3
  处理逻辑4 --> 消息队列3

消息系统与传统的数据库有着本质的区别：数据临时与永久之分

分区日志消息系统：结合了传统消息系统与数据库：既是流，又能存

每个逻辑数据源都可以建模为它自己的日志

以日志为中心的基础设施栈

流与数据库

本质上就是状态复制机的实现

本质上数据库的日志就是流，数据库里的数据就是当前流重放的快照

DataFlow模型

核心概念：

时间问题

流处理依赖于本地时间戳，时钟是不可靠的，同时考虑消息堆积、软件错误等问题，基于时间戳的流分析可能不准

窗口类型

  1. 轮转窗口：固定长度，相互之间没有重叠且紧邻 [1,3] [4,6]
  2. 跳跃窗口：固定长度，允许之间重叠以进行平滑过度 [1,3] [2,4]
  3. 滑动窗口
  4. 会话窗口：没有固定时间，将同一用户的事件组合在一起

容错

这需要消费端保证幂等性，否则为了容错会输出不止一次导致副作用

特征工程

利用工程手段从“用户信息”“物品信息”“场景信息”中提取特征的过程，在已有的、可获得的数据基础上，“尽量”保留有用信息是现实中构建特征工程的原则

常用特征

  1. 用户行为数据：区分隐式反馈与显式反馈，对用户行为数据的采集与使用与业务强相关
  2. 用户关系数据：人与人之间连接的记录，区分强关系（主动建立连接）与弱关系（间接的关系导致的连接）
  3. 属性、标签类数据：物品属性、人口属性、主动打的标签等
  4. 内容类数据：描述型文字、图片，甚至视频，需要进一步通过NLP、图像识别等转为结构化信息才能作为特征使用
  5. 场景信息：描述的是用户所处的客观的推荐环境，常见的有所处于什么时空

特征处理

进行特征处理的目的，是把所有的特征全部转换成一个数值型的特征向量

类别特征处理：

One-hot 编码（也被称为独热编码），它是将类别、ID 型特征转换成数值向量的一种最典型的编码方式。它通过把所有其他维度置为 0，单独将当前类别或者 ID 对应的维度置为 1 的方式生成特征向量

周二 => [0,1,0,0,0,0,0] -- 将一周7天视为7个维度，将周二所在的维度设为1

数值类特征处理：

  1. 归一化
  2. 分桶：将样本按照某特征的值从高到低排序，然后按照桶的数量找到分位数，将样本分到各自的桶中，再用桶 ID 作为特征值
分桶：
[1,1,1,1,1,1,5,8,10] => [(1,5),(5,10)]

Embedding

用一个数值向量“表示”一个对象（Object）的方法

词 Embedding：

生成 Skip-gram 模型结构的训练数据

在通过神经网络训练得到模型，一个词就可以通过模型推断，转为向量

图 Embedding：

  1. Deep Walk：在由物品组成的图结构上进行随机游走，产生大量物品序列，然后将这些物品序列作为训练样本输入 Word2vec 进行训练，最终得到物品的 Embedding
  2. Node2vec：通过调整随机游走跳转概率的方法，让 Graph Embedding 的结果在网络的同质性（Homophily）和结构性（Structural Equivalence）中进行权衡。同质性指的是距离相近节点的 Embedding 应该尽量近似，结构性指的是结构上相似的节点的 Embedding 应该尽量接近
    1. 为了使 Graph Embedding 的结果能够表达网络的“结构性”，在随机游走的过程中，需要让游走的过程更倾向于 BFS（Breadth First Search，广度优先搜索），因为 BFS 会更多地在当前节点的邻域中进行游走遍历，相当于对当前节点周边的网络结构进行一次“微观扫描”。当前节点是“局部中心节点”，还是“边缘节点”，亦或是“连接性节点”，其生成的序列包含的节点数量和顺序必然是不同的，从而让最终的 Embedding 抓取到更多结构性信息
    2. 而为了表达“同质性”，随机游走要更倾向于 DFS（Depth First Search，深度优先搜索）才行，因为 DFS 更有可能通过多次跳转，游走到远方的节点上。但无论怎样，DFS 的游走更大概率会在一个大的集团内部进行，这就使得一个集团或者社区内部节点的 Embedding 更为相似，从而更多地表达网络的“同质性”

Embedding 可以直接使用，在到 Embedding 向量之后，直接利用 Embedding 向量的相似性实现某些推荐系统的功能。也可以预先训练好物品和用户的 Embedding 之后，不直接应用，而是把这些 Embedding 向量作为特征向量的一部分，跟其余的特征向量拼接起来，作为推荐模型的输入参与训练。最后是一种 E2E 的应用，即不预先训练 Embedding，而是把 Embedding 的训练与深度学习推荐模型结合起来，采用统一的、端到端的方式一起训练，直接得到包含 Embedding 层的推荐模型。

非负矩阵因式分解

输入多个样本数据，每个样本数据都是一个m维数值向量，首先把我们的数据集用矩阵的形式写出来，每一列是一个数据，而每一行是这些数据对应维度的数值。于是我们就有了一个大小为m*n的输入矩阵。而算法的目标就是将这个矩阵分解为另外两个非负矩阵的积

模式

Workflow

复制模式

flowchart
  数据集 --> 复制器
  复制器 --> 工作流1
  复制器 --> 工作流2
  复制器 --> 工作流3

过滤模式

flowchart
  数据集(1,2,3) --> 过滤器
  过滤器 --> 工作流(1,2)

分离模式

flowchart
  数据集(1,2,3) --> 分离器
  分离器 --> 工作流1(1,2)
  分离器 --> 工作流2(2,3)

合并模式

flowchart
  数据集1(1,2) --> 合并器
  数据集2(2,3) --> 合并器
  合并器 --> 工作流(1,2,3)

发布订阅

架构

Lambda

完整的数据集 = λ (实时数据) * λ (历史数据)

stateDiagram-v2
  数据 --> 批处理层
  数据 --> 速度层
  state 服务层 {
    批处理数据
    速度数据
  }
  速度层 --> 速度数据
  批处理层 --> 批处理数据

  应用 --> 批处理数据
  应用 --> 速度数据

Kappa

stateDiagram-v2
  state 速度层 {
    任务N
    任务N+1
  }
  数据 --> 速度层
  state 服务层 {
    数据N
    数据N+1
  }
  任务N --> 数据N
  任务N+1 --> 数据N+1
  应用 --> 数据N+1

Spark

Spark 比 MapReduce 快的原因：更为简单的 RDD 编程模型减少了作业调度次数，以及优先使用内存

各组件

  1. SparkContext 启动 DAGScheduler 构造执行的 DAG 图，拆分成计算任务
  2. Driver 向 Cluster Manager 请求计算资源，分配 Worker
  3. Worker 向 Driver 注册并下载代码执行

RDD

分区：同一个 RDD 包含的数据被存储在系统的不同节点中，需要读取时根据ID 和分区的 index 可以唯一确定对应数据块的编号，从而通过底层存储层的接口中提取到数据进行处理

不可变：一个 RDD 都是只读的，只可以对现有的 RDD 进行转换（Transformation）操作，得到新的 RDD 作为中间计算的结果

并行：由于上面两个特性，就可以并行对 RDD 进行操作

结构

SparkContext：所有 Spark 功能的入口，它代表了与 Spark 节点的连接，一个线程只有一个 SparkContext

SparkConf： 一些参数配置信息

Partitions：数据的逻辑结构，每个 Partition 会映射到某个节点内存或硬盘的一个数据块

Dependencies：每一步产生的 RDD 里都会存储它的依赖关系，即它是通过哪个 RDD 经过哪个转换操作得到的

窄依赖，父 RDD 的分区可以一一对应到子 RDD 的分区

宽依赖，父 RDD 的每个分区可以被多个子 RDD 的分区使用

窄依赖允许子 RDD 的每个分区可以被并行处理产生，而宽依赖则必须等父 RDD 的所有分区都被计算好之后才能开始处理

Checkpoint：对于一些计算过程比较耗时的 RDD，可以进行持久化，标记这个 RDD 有被检查点处理过，并且清空它的所有依赖关系，这样在进行崩溃恢复的时候就不用在向前向父 RDD 回溯

Storage Level：记录 RDD 持久化时的存储级别，内存或内存硬盘 或在分区节点上内存、内存硬盘

Iterator：迭代函数，Compute：计算函数 都是用来表示 RDD 怎样通过父 RDD 计算得到的

数据操作

大部分操作跟Stream差不多

Spark 的 Shuffle 操作跟 MapReduce 是一样的，其通过生产与消费 Shuffle 中间文件的方式，来完成集群范围内的数据交换

Shuffle 中间文件

调度系统

  1. DAGScheduler 以 Shuffle 为边界，将开发者设计的计算图 DAG 拆分为多个执行阶段 Stages，然后为每个 Stage 创建任务集 TaskSet
  2. SchedulerBackend 通过与 Executors 中的 ExecutorBackend 的交互来实时地获取集群中可用的计算资源，并将这些信息记录到 ExecutorDataMap 数据结构
  3. 与此同时，SchedulerBackend 根据 ExecutorDataMap 中可用资源创建 WorkerOffer，以 WorkerOffer 为粒度提供计算资源
  4. 对于给定 WorkerOffer，TaskScheduler 结合 TaskSet 中任务的本地性倾向，按照 PROCESS_LOCAL、NODE_LOCAL、RACK_LOCAL 和 ANY 的顺序，依次对 TaskSet 中的任务进行遍历，优先调度本地性倾向要求苛刻的 Task
  5. 被选中的 Task 由 TaskScheduler 传递给 SchedulerBackend，再由 SchedulerBackend 分发到 Executors 中的 ExecutorBackend。Executors 接收到 Task 之后，即调用本地线程池来执行分布式任务。

存储系统

SparkSQL

架构

DataSet

DataSet 所描述的数据都被组织到有名字的列中，就像关系型数据库中的表一样

20221218152850

DataFrame

可以被看作是一种特殊的 DataSet，但是它的每一列并不存储类型信息，所以在编译时并不能发现类型错误

SparkStreaming

Spark Streaming 用时间片拆分了无限的数据流，然后对每一个数据片用类似于批处理的方法进行处理，输出的数据也是一块一块的，通过提供了一个对于流数据的抽象 DStream 来描述数据流，底层 DStream 也是由很多个序列化的 RDD 构成，按时间片（比如一秒）切分成的每个数据单位都是一个 RDD

20221218153730

主要缺点是实时计算延迟较高，这是由于 Spark Streaming 不支持太小的批处理的时间间隔

StructuredStreaming

输入的数据流按照时间间隔（以一秒为例）划分成数据段。每一秒都会把新输入的数据添加到表中，Spark 也会每秒更新输出结果。输出结果也是表的形式，输出表可以写入硬盘或者 HDFS。

20221218154712

Structured Streaming 提供一个 level 更高的 API，这样的数据抽象可以让开发者用一套统一的方案去处理批处理和流处理

相比 SparkStreaming，StructuredStreaming可以支持更小的时间间隔，2.3 也引入了连续处理模式，同时也有对事件时间的支持

架构

20221219144426

核心模型

最核心的数据结构是 Stream，它代表一个运行在多个分区上的并行流

当一个 Flink 程序被执行的时候，它会被映射为 Streaming Dataflow：

2022121914389

程序天生是并行和分布式的。一个 Stream 可以包含多个分区（Stream Partitions），一个操作符可以被分成多个操作符子任务，每一个子任务是在不同的线程或者不同的机器节点中独立执行的：

20221219144240

Beam

一个适配流处理、批处理的中间层

编程模型

PCollection

特性：

Transform

stateDiagram-v2
  数据1 --> Transform
  数据2 --> Transform
  Transform --> 数据3
  Transform --> 数据4

常见的 Transform 接口：

Pipeline

stateDiagram-v2
  输入 --> PCollection1: Transform1
  PCollection1 --> PCollection2: Transform2
  PCollection2 --> PCollection3: Transform3
  PCollection3 --> 输出: Transform4

分布式环境下，整个数据流水线会启动 N 个 Workers 来同时处理 PCollection，在具体处理某一个特定 Transform 的时候，数据流水线会将这个 Transform 的输入数据集 PCollection 里面的元素分割成不同的 Bundle，将这些 Bundle 分发给不同的 Worker 来处理

在单个 Transfrom中，如果某一个 Bundle 里面的元素因为任意原因导致处理失败了，则这整个 Bundle 里的元素都必须重新处理

在多步骤的 Transform 上，如果处理的一个 Bundle 元素发生错误了，则这个元素所在的整个 Bundle 以及与这个 Bundle 有关联的所有 Bundle 都必须重新处理

IO

StreamingSQL

/* 窗口：最近10个温度的平均值 */
Select bid, avg(t) as T From BoilerStream WINDOW HOPPING (SIZE 10, ADVANCE BY 1);
/* join */
from TempStream[temp > 30.0]#window.time(1 min) as T
  join RegulatorStream[isOn == false]#window.length(1) as R
  on T.roomNo == R.roomNo
select T.roomNo, R.deviceID, 'start' as action
insert into RegulatorActionStream; // Siddhi Streaming SQL
/*  某个模式有没有在特定的时间段内发生 */
from every( e1=TempStream ) -> e2=TempStream[ e1.roomNo == roomNo and (e1.temp + 5) <= temp ]
    within 10 min
select e1.roomNo, e1.temp as initialTemp, e2.temp as finalTemp
insert into AlertStream;