spark 中的RDD编程 -以下基于Java api

JavaRDD<String> lines=sc.textFile(inputFile);

List<String> list=new ArrayList<String>();
//建立列表，列表中包含以下自定义表项
list.add("error:a");
list.add("error:b");
list.add("error:c");
list.add("warning:d");
list.add("hadppy ending!");
//将列表转换为RDD对象
JavaRDD<String> lines = sc.parallelize(list);
//将RDD对象lines中有error的表项过滤出来，放在RDD对象errorLines中
JavaRDD<String> errorLines = lines.filter(
        new Function<String, Boolean>() {
            public Boolean call(String v1) throws Exception {
                return v1.contains("error");
            }
        }
);
//遍历过滤出来的列表项
List<String> errorList = errorLines.collect();
for (String line : errorList)
    System.out.println(line);

/**
 * Take the first num elements of the RDD. This currently scans the partitions *one by one*, so
 * it will be slow if a lot of partitions are required. In that case, use collect() to get the
 * whole RDD instead.
 */
def take(num: Int): JList[T] 

JavaRDD<String> unionLines=errorLines.union(warningLines);

for(String line :unionLines.take(2))
    System.out.println(line);

List<String> unions=unionLines.collect();
for(String line :unions)
    System.out.println(line);

JavaRDD<String> errorLines=lines.filter(
        new Function<String, Boolean>() {
            public Boolean call(String v1)throws Exception {
                return v1.contains("error");
            }
        }
);

List<String> strLine=new ArrayList<String>();
strLine.add("how are you");
strLine.add("I am ok");
strLine.add("do you love me")
JavaRDD<String> input=sc.parallelize(strLine);
JavaRDD<String> words=input.flatMap(
        new FlatMapFunction<String, String>() {
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" "));
            }
        }
);

class ContainError implements Function<String,Boolean>{
    public Boolean call(String v1) throws Exception {
        return v1.contains("error");
    }
}
JavaRDD<String> errorLines=lines.filter(new ContainError());
for(String line :errorLines.collect())
    System.out.println(line);

JavaRDD<Integer> rdd =sc.parallelize(Arrays.asList(1,2,3,4));
JavaRDD<Integer> result=rdd.map(
    new Function<Integer, Integer>() {
        public Integer call(Integer v1) throwsException {
        return v1*v1;
        }
    }
);
System.out.println( StringUtils.join(result.collect(),","));

JavaRDD<Integer> rdd =sc.parallelize(Arrays.asList(1,2,3,4));
JavaRDD<Integer> results=rdd.filter(
new Function<Integer, Boolean>() {
    public Boolean call(Integer v1) throws Exception {
        return v1!=1;
        }
    }
);
System.out.println(StringUtils.join(results.collect(),","));

JavaRDD<String> rdd =sc.parallelize(Arrays.asList("hello world","hello you","world i love you"));
JavaRDD<String> words=rdd.flatMap(
    new FlatMapFunction<String, String>() {
        public Iterable<String> call(String s) throws Exception {
            return Arrays.asList(s.split(" "));
        }
    }
);
System.out.println(StringUtils.join(words.collect(),‘\n‘));

JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1,2));
JavaRDD<Integer> rdd2 = sc.parallelize(Arrays.asList(1,2));
JavaPairRDD<Integer ,Integer> rdd=rdd1.cartesian(rdd2);
for(Tuple2<Integer,Integer> tuple:rdd.collect())
    System.out.println(tuple._1()+"->"+tuple._2());

JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4,5,6,7,8,9,10));
Integer sum =rdd.reduce(
    new Function2<Integer, Integer, Integer>() {
        public Integercall(Integer v1, Integer v2) throws Exception {
            return v1+v2;
        }
    }
);
System.out.println(sum.intValue());

JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4,5,6,7,8,9,10));
Integer sum =rdd.fold(0,
        new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        }
);
System.out.println(sum);

JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4,5,6,7,8,9,10));
Integer result =rdd.fold(1,
        new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1*v2;
            }
        }
);
System.out.println(result);

public class AvgCount implements Serializable{
public int total;
    public int num;
    public AvgCount(int total,int num){
    this.total=total;
    this.num=num;
}
public double avg(){
    return total/(double)num;
}
static Function2<AvgCount,Integer,AvgCount> addAndCount=
new Function2<AvgCount, Integer, AvgCount>() {
    public AvgCount call(AvgCount a, Integer x) throws Exception {
        a.total+=x;
        a.num+=1;
        return a;
        }
};
static Function2<AvgCount,AvgCount,AvgCount> combine=
    new Function2<AvgCount, AvgCount, AvgCount>() {
        public AvgCount call(AvgCount a, AvgCount b) throws Exception {
            a.total+=b.total;
            a.num+=b.num;
            return a;
        }
 };
    public static void main(String args[]){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("my app");
        JavaSparkContext sc = new JavaSparkContext(conf);

        AvgCount intial =new AvgCount(0,0);
        JavaRDD<Integer> rdd =sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6));
        AvgCount result=rdd.aggregate(intial,addAndCount,combine);
        System.out.println(result.avg());

    }

}

JavaRDD<Integer> rdd =sc.parallelize(Arrays.asList(1,2,3,4,5));
rdd.persist(StorageLevel.MEMORY_ONLY());
System.out.println(rdd.count());
System.out.println(StringUtils.join(rdd.collect(),‘,‘));