Spark编程进阶

SparkConf conf =new SparkConf().setMaster("local").setAppName("analysis");
JavaSparkContext sc =new JavaSparkContext(conf);
JavaRDD<String> rdd=sc.textFile("c:\\tests.txt");
final Accumulator<Integer> blankLines =sc.accumulator(0);
JavaRDD<String> callSigns = rdd.flatMap(
        new FlatMapFunction<String, String>() {

            public Iterable<String> call(String s) throws Exception {
                if (s.equals(""))
                    blankLines.add(1);
                return Arrays.asList(s.split(" "));
            }

        }
);  
callSigns.saveAsTextFile("output.txt");
System.out.println("Blank Lines: "+blankLines.value());

//广播变量，每个的国家区号表
final Broadcast<String[]> signPrefixes =sc.broadcast(loadCallSignTable());
JavaPairRDD<String,Integer> countryContactCounts =contactCounts.mapToPair(
    new PairFunction<Tuple2<String,Integer>,String,Integer>(){
        public Tuple2<String,Integer> call<Tuple2<String,Integer>callSignCount){
            //获取区号前缀
            String sign=callSignCount._1();
            //获取该前缀，和广播变量比较，得到其对应的国家
            String country=lookupCountry(sign,signPrefixed.value());
            //将(电话，通信次数)转换为(国家，通信次数)PairRDD
            return new Tuple2(country,callSignCount._2());
        }
    }//针对同一个国家做归并，求每一个国家的通信次数之和
).reduceByKey(new SumInts());
//保存结果
contryContactCounts.saveAsTextFile(outputDir+"/countrys.txt")

 /*我们有一个在线的业余呼号数据库，可以用这个数据库查询日志中级路过的联系人呼号列表。可以通过基于分区的操作，在每个分区中共享一个数据库连接池*/
 JavaPairRDD<String,CallLog[]>contactsContactLists =
    validCallSigns.mapPartitionsToPair(
        new PairFlatMapFunction<Iterator<String>,String,CallLog[]>(){
                public Iterable<Tuple2<String,CallLog[]>> call(Iterator<String> input){
                    ArrayList<Tuple2<String,CallLog[]>> callSignLogs =new ArrayList<Tuple2<String, CallLog[]>>();
                    Arraylist<Tuple2<String,ContentExchange>> requests =new ArrayList<Tuple2<String,ContentExchange>>();
                    ObjectMapper mapper=createMapper();
                    HttpClient client =new HttpClient();
                    try{
                            client.start();
                            while(input.hasNext()){
                                requests.add(createRequestForSign(input.next(),client));
                            }
                            for(Tuple2<String,ContentExchange> signExchange :requests){
                                callSignLogs.add(fetchResultFromRequest(mapper,signExchange));
                            }
                       }catch(Exception e){
                            e.printStackTrace();
                    }
                    return callSignLogs;
                }
            }
 );

/** Add a value into this StatCounter, updating the internal statistics. */
  def merge(value: Double): StatCounter = {//此处mu是平均值,m2是每个数与平均数的差的平方和
    val delta = value - mu
    n += 1
    mu += delta / n
    m2 += delta * (value - mu)
    maxValue = math.max(maxValue, value)
    minValue = math.min(minValue, value)
    this
  }