30 Apache Beam实战冲刺:Beam如何run everywhere_ 你好,我是蔡元楠。
今天我要与你分享的主题是“Apache Beam实战冲刺:Beam如何run everywhere”。
你可能已经注意到,自第26讲到第29讲,从Pipeline的输入输出,到Pipeline的设计,再到Pipeline的测试,Beam Pipeline的概念一直贯穿着文章脉络。那么这一讲,我们一起来看看一个完整的Beam Pipeline究竟是如何编写的。
Beam Pipeline
一个Pipeline,或者说是一个数据处理任务,基本上都会包含以下三个步骤:
- 读取输入数据到PCollection。
- 对读进来的PCollection做某些操作(也就是Transform),得到另一个PCollection。
- 输出你的结果PCollection。
这么说,看起来很简单,但你可能会有些迷惑:这些步骤具体该怎么做呢?其实这些步骤具体到Pipeline的实际编程中,就会包含以下这些代码模块:
Java
// Start by defining the options for the pipeline. PipelineOptions options = PipelineOptionsFactory.create(); // Then create the pipeline. Pipeline pipeline = Pipeline.create(options); PCollection
从上面的代码例子中你可以看到,第一行和第二行代码是创建Pipeline实例。任何一个Beam程序都需要先创建一个Pipeline的实例。Pipeline实例就是用来表达Pipeline类型的对象。这里你需要注意,一个二进制程序可以动态包含多个Pipeline实例。
还是以之前的美团外卖电动车处理的例子来做说明吧。
比如,我们的程序可以动态判断是否存在第三方的电动车图片,只有当有需要处理图片时,我们才去创建一个Pipeline实例处理。我们也可以动态判断是否存在需要转换图片格式,有需要时,我们再去创建第二个Pipeline实例。这时候你的二进制程序,可能包含0个、1个,或者是2个Pipeline实例。每一个实例都是独立的,它封装了你要进行操作的数据,和你要进行的操作Transform。
Pipeline实例的创建是使用Pipeline.create(options)这个方法。其中options是传递进去的参数,options是一个PipelineOptions这个类的实例。我们会在后半部分展开PipelineOptions的丰富变化。
第三行代码,我们用TextIO.read()这个Transform读取了来自外部文本文件的内容,把所有的行表示为一个PCollection。
第四行代码,用 lines.apply(new FilterLines()) 对读进来的PCollection进行了过滤操作。
第五行代码 filteredLines.apply(“WriteMyFile”, TextIO.write().to(“gs://some/outputData.txt”)),表示把最终的PCollection结果输出到另一个文本文件。
程序运行到第五行的时候,是不是我们的数据处理任务就完成了呢?并不是。
记得我们在第24讲、第25讲中提过,Beam是延迟运行的。程序跑到第五行的时候,只是构建了Beam所需要的数据处理DAG用来优化和分配计算资源,真正的运算完全没有发生。
所以,我们需要最后一行pipeline.run().waitUntilFinish(),这才是数据真正开始被处理的语句。
这时候运行我们的代码,是不是就大功告成呢?别急,我们还没有处理好程序在哪里运行的问题。你一定会好奇,我们的程序究竟在哪里运行,不是说好了分布式数据处理吗?
在上一讲《如何测试Beam Pipeline》中我们学会了在单元测试环境中运行Beam Pipeline。就如同下面的代码。和上文的代码类似,我们把Pipeline.create(options)替换成了TestPipeline.create()。
Java
Pipeline p = TestPipeline.create(); PCollection
TestPipeline是Beam Pipeline中特殊的一种,让你能够在单机上运行小规模的数据集。之前我们在分析Beam的设计理念时提到过,Beam想要把应用层的数据处理业务逻辑和底层的运算引擎分离开来。
现如今Beam可以做到让你的Pipeline代码无需修改,就可以在本地、Spark、Flink,或者在Google Cloud DataFlow上运行。这些都是通过Pipeline.create(options) 这行代码中传递的PipelineOptions实现的。
在实战中,我们应用到的所有option其实都是实现了PipelineOptions这个接口。
举个例子,如果我们希望将数据流水线放在Spark这个底层数据引擎运行的时候,我们便可以使用SparkPipelineOptions。如果我们想把数据流水线放在Flink上运行,就可以使用FlinkPipelineOptions。而这些都是extends了PipelineOptions的接口,示例如下:
Java options = PipelineOptionsFactory.as(SparkPipelineOptions.class); Pipeline pipeline = Pipeline.create(options);
通常一个PipelineOption是用PipelineOptionsFactory这个工厂类来创建的,它提供了两个静态工厂方法给我们去创建,分别是PipelineOptionsFactory.as(Class)和PipelineOptionsFactory.create()。像上面的示例代码就是用PipelineOptionsFactory.as(Class)这个静态工厂方法来创建的。
当然了,更加常见的创建方法是从命令行中读取参数来创建PipelineOption,使用的是PipelineOptionsFactory/#fromArgs(String[])这个方法,例如:
Java public static void main(String[] args) { PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create(); Pipeline p = Pipeline.create(options); }
下面我们来看看不同的运行模式的具体使用方法。
直接运行模式
我们先从直接运行模式开始讲。这是我们在本地进行测试,或者调试时倾向使用的模式。在直接运行模式的时候,Beam会在单机上用多线程来模拟分布式的并行处理。
使用Java Beam SDK时,我们要给程序添加Direct Runner的依赖关系。在下面这个maven依赖关系定义文件中,我们指定了beam-runners-direct-java这样一个依赖关系。
pom.xml
一般我们会把runner通过命令行指令传递进程序。就需要使用PipelineOptionsFactory.fromArgs(args)来创建PipelineOptions。PipelineOptionsFactory.fromArgs()是一个工厂方法,能够根据命令行参数选择生成不同的PipelineOptions子类。
PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
在实验程序中也可以强行使用Direct Runner。比如:
PipelineOptions options = PipelineOptionsFactory.create(); options.setRunner(DirectRunner.class); // 或者这样 options = PipelineOptionsFactory.as(DirectRunner.class); Pipeline pipeline = Pipeline.create(options);
如果是在命令行中指定Runner的话,那么在调用这个程序时候,需要指定这样一个参数–runner=DirectRunner。比如:
mvn compile exec:java -Dexec.mainClass=YourMainClass \ -Dexec.args=”–runner=DirectRunner” -Pdirect-runner
Spark运行模式
如果我们希望将数据流水线放在Spark这个底层数据引擎运行的时候,我们便可以使用Spark Runner。Spark Runner执行Beam程序时,能够像原生的Spark程序一样。比如,在Spark本地模式部署应用,跑在Spark的RM上,或者用YARN。
Spark Runner为在Apache Spark上运行Beam Pipeline提供了以下功能:
- Batch 和streaming的数据流水线;
- 和原生RDD和DStream一样的容错保证;
- 和原生Spark同样的安全性能;
- 可以用Spark的数据回报系统;
- 使用Spark Broadcast实现的Beam side-input。
目前使用Spark Runner必须使用Spark 2.2版本以上。
这里,我们先添加beam-runners-spark的依赖关系。
然后,要使用SparkPipelineOptions传递进Pipeline.create()方法。常见的创建方法是从命令行中读取参数来创建PipelineOption,使用的是PipelineOptionsFactory.fromArgs(String[])这个方法。在命令行中,你需要指定runner=SparkRunner:
mvn exec:java -Dexec.mainClass=YourMainClass \ -Pspark-runner \ -Dexec.args=”–runner=SparkRunner \ –sparkMaster=
也可以在Spark的独立集群上运行,这时候spark的提交命令,spark-submit。
spark-submit –class YourMainClass –master spark://HOST:PORT target/…jar –runner=SparkRunner
当Beam程序在Spark上运行时,你也可以同样用Spark的网页监控数据流水线进度。
Flink运行模式
Flink Runner是Beam提供的用来在Flink上运行Beam Pipeline的模式。你可以选择在计算集群上比如 Yarn/Kubernetes/Mesos 或者本地Flink上运行。Flink Runner适合大规模,连续的数据处理任务,包含了以下功能:
- 以Streaming为中心,支持streaming处理和batch处理;
- 和flink一样的容错性,和exactly-once的处理语义;
- 可以自定义内存管理模型;
- 和其他(例如YARN)的Apache Hadoop生态整合比较好。
其实看到这里,你可能已经掌握了这里面的诀窍。就是通过PipelineOptions来指定runner,而你的数据处理代码不需要修改。PipelineOptions可以通过命令行参数指定。那么类似Spark Runner,你也可以使用Flink来运行Beam程序。
同样的,首先你需要在pom.xml中添加Flink Runner的依赖。
然后在命令行中指定flink runner:
mvn exec:java -Dexec.mainClass=YourMainClass \ -Pflink-runner \ -Dexec.args=”–runner=FlinkRunner \ –flinkMaster=
Google Dataflow 运行模式
Beam Pipeline也能直接在云端运行。Google Cloud Dataflow就是完全托管的Beam Runner。当你使用Google Cloud Dataflow服务来运行Beam Pipeline时,它会先上传你的二进制程序到Google Cloud,随后自动分配计算资源创建Cloud Dataflow任务。
同前面讲到的Direct Runner和Spark Runner类似,你还是需要为Cloud Dataflow添加beam-runners-google-cloud-dataflow-java依赖关系:
我们假设你已经在Google Cloud上创建了project,那么就可以用类似的命令行提交任务:
mvn -Pdataflow-runner compile exec:java \ -Dexec.mainClass=
小结
这一讲我们先总结了前面几讲Pipeline的完整使用方法。之后一起探索了Beam的重要特性,就是Pipeline可以通过PipelineOption动态选择同样的数据处理流水线在哪里运行。并且,分别展开讲解了直接运行模式、Spark运行模式、Flink运行模式和Google Cloud Dataflow运行模式。在实践中,你可以根据自身需要,去选择不同的运行模式。
思考题
Beam的设计模式是对计算引擎动态选择,它为什么要这么设计?
欢迎你把答案写在留言区,与我和其他同学一起讨论。如果你觉得有所收获,也欢迎把文章分享给你的朋友。
参考资料
https://learn.lianglianglee.com/%e4%b8%93%e6%a0%8f/%e5%a4%a7%e8%a7%84%e6%a8%a1%e6%95%b0%e6%8d%ae%e5%a4%84%e7%90%86%e5%ae%9e%e6%88%98/30%20Apache%20Beam%e5%ae%9e%e6%88%98%e5%86%b2%e5%88%ba%ef%bc%9aBeam%e5%a6%82%e4%bd%95run%20everywhere_.md
更多学习
更多实时资讯,前沿技术,生活趣事。尽在【老马啸西风】
交流社群:[交流群信息](https://mp.weixin.qq.com/s/rkSvXxiiLGjl3S-ZOZCr0Q)