摘要：要說在中常見的函數是哪一個，當然是。是一個實現了接口的抽象類，其中是數據處理方法，強制子類必須實現。以上為學習一天的總結，有錯誤歡迎指正。相同的是這個方法處理的都是中的一個元素。

在閱讀本文前，可先看一下官方的WordCount代碼， 對Apache Beam有大概的了解。

要說在Apache Beam中常見的函數是哪一個，當然是apply()。常見的寫法如下：

[Final Output PCollection] = [Initial Input PCollection].apply([First Transform])
.apply([Second Transform])
.apply([Third Transform])

而在最簡單的wordcount代碼中，就出現了許多種不同的傳入參數類型，除了輸入輸出的部分，還包括
1）使用ParDo.of():

.apply("ExtractWords-joe",
        ParDo.of(new DoFn() {
            @ProcessElement
            public void processElement(ProcessContext context) {
                System.out.println(context.element()+"~");
                for (String word : context.element().split(" ")) {
                    if (!word.isEmpty()) {
                        //輸出到Output PCollection
                        context.output(word);
                    }
                }
            }
        })
)

2）使用MapElements.via():

.apply("FomatResults",
        MapElements.via(new SimpleFunction,String>() {
            @Override
            public String apply(KV input) {
                return input.getKey()+":"+input.getValue();
            }
        }))

3）以及使用PTransform子類：

.apply(new CountWords())
  public static class CountWords extends PTransform,
      PCollection>> {
    @Override
    public PCollection> expand(PCollection lines) {

      // Convert lines of text into individual words.
      PCollection words = lines.apply(
          ParDo.of(new ExtractWordsFn()));

      // Count the number of times each word occurs.
      PCollection> wordCounts =
          words.apply(Count.perElement());

      return wordCounts;
    }
  }

這么多種傳入方式到底有什么聯系？通過查看源碼可以看出apply函數的定義如下：

  public  OutputT apply(
      String name, PTransform root) {
    return begin().apply(name, root);
  }

傳入的參數為PTransform類對象，也就是這幾種傳入參數其實都是PTransform類的變形。
PTransform是一個實現了Serializable接口的抽象類，其中public abstract OutputT expand(InputT input); 是數據處理方法，強制子類必須實現。
因此第(3)種方式很容易理解，就是通過繼承PTransform并實現了expand方法定義了CountWords類，給apply方法傳遞了一個CountWords對象。

在第(2)種方式中，MapElements是PTransform的子類，實現了expand方法，其實現方式是調用@Nullable private final SimpleFunction fn;成員中定義的數據處理方法，MapElements.via()則是一個為初始化fn的靜態方法，定義如下：

  public static  MapElements via(
      final SimpleFunction fn) {
    return new MapElements<>(fn, null, fn.getClass());
  }

傳入了一個SimpleFunction對象，SimpleFunction是一個必須實現public OutputT apply(InputT input) 方法的抽象類，用戶在該apply方法中實現數據處理。
所以這種方式的實現方式如下：
定義SimpleFunction的子類并實現其中的apply方法，將該子類的對象傳遞給MapElements.via()。

第(1)種方式中，ParDo.of()方法傳入一個DoFn對象， 返回一個SingleOutput對象：

  public static  SingleOutput of(DoFn fn) {
    validate(fn);
    return new SingleOutput(
        fn, Collections.>emptyList(), displayDataForFn(fn));
  }

SingleOutput與MapElements類似，也是PTransform的子類，實現了expand方法，使用private final DoFn fn;成員中的方法進行數據處理。
而DoFn是一個抽象類，用戶必須實現其注解方法(存疑) public void processElement(ProcessContext c)。
所以這種方式的實現方式如下：
定義DoFn的子類并實現其中的processElement方法，將該子類的對象傳遞給ParDo.of()。
需要注意的是processElement方法與前2種方式不同，輸入和輸出數據都是在傳入參數ProcessContext c中，而不是通過return進行傳遞。

以上為學習Apache Beam一天的總結，有錯誤歡迎指正。

**

**
1）MapElement.via(SimpleFunction)和PTransform
MapElements是PTransform的一個子類：
public class MapElements
extends PTransform, PCollection>
從泛型參數來看，PTransform處理的是PCollection，而MapElement處理的是PCollection中的一個元素，對比SimpleFunction的apply方法和PTransform的expand方法的實現方式得到驗證。

2）MapElement.via(SimpleFunction)和ParDo.of(DoFn)
區別之前已經說過，DoFn的processElement方法的輸入和輸出都是從參數傳入，而SimpleFunction的apply方法從參數傳入輸入，從return傳出輸出。
相同的是這2個方法處理的都是PCollection中的一個元素。
查看MapElement的expand方法源碼：

@Override
public PCollection expand(PCollection input) {
  checkNotNull(fn, "Must specify a function on MapElements using .via()");
  return input.apply(
      "Map",
      ParDo.of(
          new DoFn() {
            @ProcessElement
            public void processElement(ProcessContext c) {
              c.output(fn.apply(c.element()));
            }
    //部分代碼忽略
          }));
}

可以看出其實也是實現了DoFn的子類，在DoFn的processElement方法中調用SimpleFunction對象的apply方法進行處理。