十年网站开发经验 + 多家企业客户 + 靠谱的建站团队
量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决
成都创新互联公司坚持“要么做到,要么别承诺”的工作理念,服务领域包括:成都网站建设、网站制作、企业官网、英文网站、手机端网站、网站推广等服务,满足客户于互联网时代的荆门网站设计、移动媒体设计的需求,帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴!
Beam刚刚开源不是很久,快2个月了。目前的版本是0.5.0版本。官方的源码中提供了4个examples.无奈这四个案例都只是WordCount的四种不同的实现。作为一个从Spark进入大数据殿堂的笔者来说,用过n多次的SparkPi的我,怎么能忍受竟然没有Pi实现的example呢。假如有了这个案例,可以非常方便的无论在开发工具中还是在集群中进行测试。于是便有了下文。笔者的文笔和技术有限。不足之处,还望朋友多多提建议。Let us come on 。
我们先来讲讲Pi的实现原理。我们是用概率统计的方法来实现的。先来想象一下,以一个单位为半径画圆,再画一个圆的外切正方形。假设一个杯子的底部就被这个正方形和正方形内切圆全部填满。做n次试验,往杯子中扔石头,落在圆内的次数除以总次数是不是Pi*r*r/2r*2r也就是Pi/4.
因此,Pi就是4倍的此概率。
以下是代码的试验。仅供参考。
可以复制代码
package org.tongfang.beam.examples;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.spark.SparkRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.*;
public class BeamPi {
public static void main(String[] args) {
// Beam Pi的自定义实现方式
//第一步骤:创建options,
//通过该对象可以选择使用哪个计算框架来计算,并且设置应用的名称
PipelineOptions options = PipelineOptionsFactory.create();
//设置job(应用)名称
options.setJobName("Beam Pi");
//设置runner为Spark
options.setRunner(SparkRunner.class);
//创建管道 p
Pipeline p = Pipeline.create(options);
//100000000次的随机试验的次数,如果资源,
//足够的大可以进行更多次的试验,用大数据的理论来说,
//理论上可以进行无数次的试验(只要不断的横向扩展计算的资源)。
List
for(int i = 0;i<1000000;i++){
list.add(i);
}
//相当于Spark从内存中读取数据,并通过map迭代访问每一个元素,
//这里迭代1000000的访问每个依次增大的数字,
//没迭代一次,做一次试验,当点落到圆内,计数增加1,否则不计数
//也就是什么也不做
//然后再近些Count计数,最后计数结果除以试验次数,就是概率。
//从数学角度来看,PI的值就是4倍这个概率。从而计算出PI的值。
p.apply(Create.of(list)).apply(ParDo.of(new DoFn
double x = 0;
double y = 0;
private static final long serialVersionUID = 1L;
@ProcessElement
public void processElement(ProcessContext c) {
x = Math.random() * 2 - 1;;
y = Math.random() * 2 - 1;;
if((x*x+y*y)<1){
c.output(1);
}else{
}
}
})).apply( Count.
apply(MapElements.via(new SimpleFunction
/**
*
*/
private static final long serialVersionUID = 1L;
public Void apply(Long input) {
Float res = (float) (4.0*(float)input / 1000000f);
System.out.println(input);
System.out.println("PI : "+res);
return null;
}
}));
//这是运行计算的关键,如果这个代码不写,
//整个代码都是懒加载,并非真正计算。
p.run().waitUntilFinish();
}
}