DAG

Transform 是 Beam 中数据处理的最基本单元

Beam 把数据转换抽象成有向图
反直觉 - PCollection 是有向图中的边，而 Transform 是有向图中的节点
- 区分节点和边的关键是看一个 Transform 是不是有一个多余的输入和输出
- 每个 Transform 都可能有大于一个的输入 PCollection，也可能输出大于一个的输出 PCollection

Apply

Beam 中的 PCollection 有一个抽象的成员函数 Apply，使用任何一个 Transform 时，都需要调用 Apply

1
2
3

final_collection = input_collection.apply(Transform1)
.apply(Transform2)
.apply(Transform3)

Transform

概述

ParDo - Parallel Do - 表达的是很通用的并行处理数据操作
GroupByKey - 把一个 Key/Value 的数据集按照 Key 归并

可以用 ParDo 实现 GroupByKey

简单实现 - 放一个全局的哈希表，然后在 ParDo 中把一个个的元素插入到该哈希表中
可能不可用，面对大规模数据时，可能无法放进一个内存哈希表
而且，PCollection 会把计算分发到不同的机器上执行

ParDo

在实际应用中，80% 的数据处理流水使用基本的 ParDo 和 DoFn

在编写 ParDo 时，输入是一个 PCollection 中的单个元素，而输出可以是 0 个、1 个、多个元素
只需要考虑好怎么处理一个元素，其余事项，Beam 会在框架层面进行优化和并行
使用 ParDo 时，需要继承它提供的 DoFn 类
- 可以将 DoFn 看作 ParDo 的一部分，ParDo 和 DoFn 是一个有机整体

static class UpperCaseFn extends DoFn<String, String> {
  @ProcessElement
  public void processElement(@Element String word, OutputReceiver<String> out) {
    out.output(word.toUpperCase());
  }
}

PCollection<String> upperCaseWords = words.apply(
    ParDo
    .of(new UpperCaseFn()));

编程界面

1	pcollection.apply(ParDo.of(new DoFn()))

Filter

挑出符合条件的元素

@ProcessElement
public void processElement(@Element T input, OutputReceiver<T> out) {
    if (IsNeeded(input)) {
      out.output(input);
    }
  }

Format

对数据集进行格式转换

@ProcessElement
  public void processElement(@Element String csvLine, OutputReceiver<tf.Example> out) {
    out.output(ConvertToTfExample(csvLine));
  }

Extract

提取数据集中的特定值（属性）

@ProcessElement
  public void processElement(@Element Item item, OutputReceiver<Integer> out) {
    out.output(item.price());
  }

Stateful Transform

Statefullness - side input/side output

简单场景都是无状态的
- 每个 DoFn 的 processElement 函数中，输出只依赖于输入
- 对应的 DoFn 类不需要维持一个成员变量
无状态的 DoFn 能保证最大的并行运算能力
- 因为 DoFn 的 processElement 可以分发到不同的机器或者不同的进程
如果 processElement 的运行需要另外的信息 - 有状态的 DoFn

static class FindUserNameFn extends DoFn<String, String> {
  @ProcessElement
  public void processElement(@Element String userId, OutputReceiver<String> out) {
    out.output(database.FindUserName(userId));
  }

  Database database;
}

因为有了共享状态（数据库连接），在使用有状态的 DoFn 时，需要格外注意 Beam 的并行特性
Beam 不仅仅会把处理函数分发到不同线程和进程，也会分发到不同的机器上执行
- 当共享数据库的读取操作时，很容易引发数据库的 QPS 过高

需要共享的状态来自于另一些 Beam 的数据处理的中间结果 - side input/side output

PCollectionView<Integer> mediumSpending = ...;

PCollection<String> usersBelowMediumSpending =
  userIds.apply(ParDo
      .of(new DoFn<String, String>() {
          @ProcessElement
          public void processElement(@Element String userId, OutputReceiver<String> out, ProcessContext c) {
            int medium = c.sideInput(mediumSpending);
            if (findSpending(userId) <= medium) {
              out.output(userId);
            }
          }
      }).withSideInputs(mediumSpending)
  );