apache beam入门之组装数据变换过程

Posted 2022-09-26

目录：apache beam 个人使用经验总结目录和入门指导（Java）

当我们通过option新建的pipeline后，需要进行一系列组装过程，最后再通过pipeline.run()启动计算。

Pipeline pipeline = Pipeline.create(option);
/**
*组装计算过程
*/
pipeline.run();

因此编写beam程序时最核心的部分就是组装计算过程,其中包含1对1转换、聚合等操作。

1对1转换(类似于map操作)

该部分通过apply函数进行装饰器模式组装， 并提供了三种方式：

apply(MapElements.via(SimpleFunction对象))

该方法用于简单的转化操作，不需要提供额外的输入参数。例子如下：

PCollection<String> pcMid = pcStart.apply(MapElements.via(new SimpleFunction<String, String>() 
    @Override
    public String apply(String input) 
        return input.toLowerCase();
    
));

可见该方法对我们而言，只要定义输入参数和return元素是如何变化的即可，不需要关注其他的东西。

apply(ParoDo.of(DoFn对象))

该方式提供了1个context来获取此刻的输入元素，并通过context进行output。 该方式能够从context获取其他信息，例如option等，但编写较麻烦，定义processElement时需要带上ProcessElement注解。

<String> pcMid = pcStart.apply(ParDo.of(new DoFn<String ,String>()
            @ProcessElement
            public void processElement(ProcessContext context) 
                // 从管道中取出的1个元素
                String inputStr = context.element();
                // 转成大写
                String outputStr = inputStr.toLowerCase();
                // 输出结果
                context.output(outputStr);
            
        ));

apply(PTransform对象)

该方式其实是对数据集PCollection的操作， 真正对元素的转化还要依赖于上面2种方法

PCollection<String> pcMid = pcStart.apply(new PTransform<PCollection<String>, PCollection<String>>() 
    @Override
    public PCollection<String> expand(PCollection<String> input) 
        return input.apply(ParDo.of(new StrToLowerCaseFn()));
    
);

注意事项：
1.使用自定义的DoFn类或者SimpleFunction类进行数据转化时，如果该类中定义了成员变量，则注意该成员必须是可序列化的（实现了Serializable接口）， 否则执行时会报序列化错误
2.不要在DoFn或SimpleFunction的转化函数中，通过1个输入元素，去生成过多的输出元素，这个过多指的是超出jvm限定内存的数量，因为对于单个input元素，output时实际上是存储到1个outputMap中，直到全部output完，才会进行下一步的处理
3.不要对输入的数据做修改，如果对输入的数据做修改，运行时会报错