简介

在文章《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 tryStagePackage INFO: 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.jar Nov 03, 2022 8:41:48 PM org.apache.beam.runners.dataflow.util.PackageUtil tryStagePackage INFO: 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