简介

在文章《Apache Beam入门及Java SDK开发初体验》中大概讲了Apapche Beam的简单概念和本地运行,本文将讲解如何把代码运行在GCP Cloud Dataflow上。

本地运行

通过maven命令来创建项目:

mvn archetype:generate \    -DarchetypeGroupId=org.apache.beam \    -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \    -DarchetypeVersion=2.37.0 \    -DgroupId=org.example \    -DartifactId=word-count-beam \    -Dversion="0.1" \    -Dpackage=org.apache.beam.examples \    -DinteractiveMode=false

上面会创建一个目录word-count-beam,里面是一个例子项目。做一些简单修改就可以使用了。

先build一次,保证依赖下载成功:

$ mvn clean package

通过IDEA本地运行一下,添加入参如下:

--output=pkslow-beam-counts --inputFile=/Users/larry/IdeaProjects/pkslow-samples/README.md

处理的文件是README.md,输出结果前缀为pkslow-beam-counts

或者通过命令行来运行也可以:

mvn compile exec:java \    -Dexec.mainClass=org.apache.beam.examples.WordCount \    -Dexec.args="--output=pkslow-beam-counts --inputFile=/Users/larry/IdeaProjects/pkslow-samples/README.md"

在GCP Cloud Dataflow上运行准备环境

要有对应的Service Account和key,当然还要有权限;

要打开对应的Service;

创建好对应的Bucket,上传要处理的文件。

运行

然后在本地执行命令如下:

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \     -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://pkslow-dataflow/temp \                  --project=pkslow --region=us-east1 \                  --inputFile=gs://pkslow-dataflow/input/README.md --output=gs://pkslow-dataflow//pkslow-counts" \     -Pdataflow-runner

日志比较长,它大概做的事情就是把相关Jar包上传到temp目录下,因为执行的时候要引用。如:

Nov 03, 2022 8:41:48 PM org.apache.beam.runners.dataflow.util.PackageUtil tryStagePackageINFO: Uploading /Users/larry/.m2/repository/org/apache/commons/commons-compress/1.8.1/commons-compress-1.8.1.jar to gs://pkslow-dataflow/temp/staging/commons-compress-1.8.1-X8oTZQP4bsxsth-9F7E31Z5WtFx6VJTmuP08q9Rpf70.jarNov 03, 2022 8:41:48 PM org.apache.beam.runners.dataflow.util.PackageUtil tryStagePackageINFO: Uploading /Users/larry/.m2/repository/org/codehaus/jackson/jackson-mapper-asl/1.9.13/jackson-mapper-asl-1.9.13.jar to gs://pkslow-dataflow/temp/staging/jackson-mapper-asl-1.9.13-dOegenby7breKTEqWi68z6AZEovAIezjhW12GX6b4MI.jar

查看Bucket,确实有一堆jar包:

接着会创建dataflow jobs开始工作了。可以查看界面的Jobs如下:

点进去可以看到流程和更多细节:

最后到Bucket查看结果也出来了:

代码

代码请看GitHub: https://github.com/LarryDpk/pkslow-samples