ODPS参考手册

/** * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package com.aliyun.odps.mapreduce.examples; import java.io.IOException; import java.math.BigDecimal; import java.util.List; import com.aliyun.odps.Record; import com.aliyun.odps.io.BooleanWritable; import com.aliyun.odps.io.LongWritable; import com.aliyun.odps.io.TableInfo; import com.aliyun.odps.io.TableOutputFormat; import com.aliyun.odps.mapreduce.JobClient; import com.aliyun.odps.mapreduce.JobConf; import com.aliyun.odps.mapreduce.MapContext; import com.aliyun.odps.mapreduce.Mapper; import com.aliyun.odps.mapreduce.OdpsConf; import com.aliyun.odps.mapreduce.ReduceContext; import com.aliyun.odps.mapreduce.Reducer; import com.aliyun.odps.util.OdpsConnectionFactory; import com.aliyun.openservices.ClientException; import com.aliyun.openservices.odps.ODPSConnection; import com.aliyun.openservices.odps.ODPSException; import com.aliyun.openservices.odps.Project; import com.aliyun.openservices.odps.tables.RecordReader; import com.aliyun.openservices.odps.tables.Table; /** * A Map-reduce program to estimate the value of Pi using quasi-Monte Carlo * method. * * Mapper: Generate points in a unit square and then count points inside/outside * of the inscribed circle of the square. * * Reducer: Accumulate points inside/outside results from the mappers. * * Let numTotal = numInside + numOutside. The fraction numInside/numTotal is a * rational approximation of the value (Area of the circle)/(Area of the * square), where the area of the inscribed circle is Pi/4 and the area of unit * square is 1. Then, Pi is estimated value to be 4(numInside/numTotal). */ public class PiEstimator { /** * 2-dimensional Halton sequence {H(i)}, where H(i) is a 2-dimensional point * and i >= 1 is the index. Halton sequence is used to generate sample points * for Pi estimation. */ private static class HaltonSequence { /** Bases */ static final int[] P = { 2, 3 }; /** Maximum number of digits allowed */ static final int[] K = { 63, 40 }; private long index; private double[] x; private double[][] q; private int[][] d; /** * Initialize to H(startindex), so the sequence begins with H(startindex+1). */ HaltonSequence(long startindex) { index = startindex; x = new double[K.length]; q = new double[K.length][]; d = new int[K.length][]; for (int i = 0; i < K.length; i++) { q[i] = new double[K[i]]; d[i] = new int[K[i]]; } for (int i = 0; i < K.length; i++) { long k = index; x[i] = 0; for (int j = 0; j < K[i]; j++) { q[i][j] = (j == 0 ? 1.0 : q[i][j - 1]) / P[i]; d[i][j] = (int) (k % P[i]); k = (k - d[i][j]) / P[i]; x[i] += d[i][j] * q[i][j]; } } } /** * Compute next point. Assume the current point is H(index). Compute * H(index+1). * * @return a 2-dimensional point with coordinates in [0,1)^2 */ double[] nextPoint() { index++; for (int i = 0; i < K.length; i++) { for (int j = 0; j < K[i]; j++) { d[i][j]++; x[i] += q[i][j]; if (d[i][j] < P[i]) { break; } d[i][j] = 0; x[i] -= (j == 0 ? 1.0 : q[i][j - 1]); } } return x; } } /** * Mapper class for Pi estimation. Generate points in a unit square and then * count points inside/outside of the inscribed circle of the square. */ public static class PiMapper extends Mapper<BooleanWritable, LongWritable> { private long offset = 0L; private long size = 0L; @Override protected void setup(MapContext<BooleanWritable, LongWritable> context) { JobConf conf = (JobConf) context.getConfiguration(); long numPoints = getNumPoints(conf); int id = context.getTaskAttemptID().getTaskId(); offset = id * numPoints; size = numPoints; } /** * Map method. * * @param offset * samples starting from the (offset+1)th sample. * @param size * the number of samples for this map * @param out * output {ture->numInside, false->numOutside} * @param reporter */ @Override public void run(MapContext<BooleanWritable, LongWritable> context) throws IOException { setup(context); final HaltonSequence haltonsequence = new HaltonSequence(offset); long numInside = 0L; long numOutside = 0L; for (long i = 0; i < size;) { // generate points in a unit square final double[] point = haltonsequence.nextPoint(); // count points inside/outside of the inscribed circle of the square final double x = point[0] - 0.5; final double y = point[1] - 0.5; if (x * x + y * y > 0.25) { numOutside++; } else { numInside++; } // report status i++; if (i % 1000 == 0) { context.setStatus("Generated " + i + " samples."); } } // output map results context.write(new BooleanWritable(true), new LongWritable(numInside)); context.write(new BooleanWritable(false), new LongWritable(numOutside)); } } /** * Reducer class for Pi estimation. Accumulate points inside/outside results * from the mappers. */ public static class PiReducer extends Reducer<BooleanWritable, LongWritable> { private long numInside = 0L; private long numOutside = 0L; /** * Accumulate number of points inside/outside results from the mappers. * * @param isInside * Is the points inside? * @param values * An iterator to a list of point counts * @param output * dummy, not used here. * @param reporter */ @Override public void reduce(BooleanWritable isInside, Iterable<LongWritable> values, ReduceContext<BooleanWritable, LongWritable> context) throws IOException { if (isInside.get()) { for (LongWritable value : values) numInside += value.get(); } else { for (LongWritable value : values) numOutside += value.get(); } } /** * Reduce task done, write output to a file. */ @Override public void cleanup(ReduceContext<BooleanWritable, LongWritable> context) throws IOException { Record outputRecord = context.createOutputRecord(); outputRecord.set(0, new LongWritable(numInside)); outputRecord.set(1, new LongWritable(numOutside)); context.write(outputRecord); } } private static void setNumPoints(JobConf conf, long numPoints) { conf.setLong("num.points", numPoints); } private static long getNumPoints(JobConf conf) { return conf.getLong("num.points", 100); } /** * Run a map/reduce job for estimating Pi. * * @return the estimated value of Pi * @throws ClientException * @throws ODPSException */ public static BigDecimal estimate(int numMaps, long numPoints, JobConf jobConf, String prjName, String tableOut) throws IOException, ClientException, ODPSException { // setup