Spark 의 combineByKey 상세 설명 자바
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import java.util.ArrayList;
import java.util.List;
/**
* Created by hadoop on 17-10-23.
*/
public class JavaSparkCombine {
public static void main(String[]args){
SparkConf conf=new SparkConf().setAppName("join").setMaster("local");
JavaSparkContext sc=new JavaSparkContext(conf);
List,Integer>> users=new ArrayList,Integer>>();
Tuple2,Integer> user1=new Tuple2,Integer>("1212",1);
Tuple2,Integer> user2=new Tuple2,Integer>("1213",3);
Tuple2,Integer> user3=new Tuple2,Integer>("1214",6);
Tuple2,Integer> user4=new Tuple2,Integer>("1215",3);
Tuple2,Integer> user5=new Tuple2,Integer>("1212",1);
users.add(user1);
users.add(user2);
users.add(user3);
users.add(user4);
users.add(user5);
JavaPairRDD,Integer> userrdd=sc.parallelizePairs(users,2);
JavaRDD,Integer>> userr=sc.parallelize(users);
JavaPairRDD,Tuple2,Integer>>combrdd=userrdd.combineByKey(
new Function,Tuple2,Integer>>(){
public Tuple2,Integer> call(Integer x){
return new Tuple2,Integer> (x,1);
}
}, new Function2,Integer>,Integer,Tuple2,Integer>>(){
public Tuple2,Integer> call(Tuple2,Integer> v,Integer x){
return new Tuple2,Integer>(v._1()+x,v._2()+1);
}
},
new Function2,Integer>,Tuple2,Integer>,Tuple2,Integer>>(){
public Tuple2,Integer> call(Tuple2,Integer> v,Tuple2,Integer> v1){
return new Tuple2,Integer>(v._1()+v1._1(),v._2()+v1._2());
}
});
//groupbykey
JavaPairRDD,ArrayList>combrddgroupby=userrdd.combineByKey(
new Function, ArrayList>(){
public ArrayList call(Integer x){
List result=new ArrayList();
result.add(x);
return (ArrayList) result;
}
}, new Function2,Integer,ArrayList>(){
public ArrayList call(ArrayList v,Integer x){
v.add(x);
return v;
}
},
new Function2,ArrayList,ArrayList>(){
public ArrayList call(ArrayList v1,ArrayList v2){
// for(Integer value:v1){
// v2.add(value) ;
// }
v1.forEach(v2::add);//
return v2;
}
});
//combrddgroupby.collect().forEach(System.out::println);
//reducebykey
JavaPairRDD,Tuple2,Integer>> reducerd=userr.mapToPair(
new PairFunction, Integer>, String, Tuple2, Integer>>() {
@Override
public Tuple2, Tuple2, Integer>> call(Tuple2, Integer> s) throws Exception {
return new Tuple2, Tuple2, Integer>>(s._1(),new Tuple2,Integer>(s._2(),1));
}
}
);
JavaPairRDD,Tuple2,Integer>> reducBykey=reducerd.reduceByKey(
new Function2,Integer>,Tuple2,Integer>,Tuple2,Integer>>(){
@Override
public Tuple2, Integer> call(Tuple2, Integer> integerIntegerTuple2, Tuple2, Integer> integerIntegerTuple22) throws Exception {
return new Tuple2,Integer>(integerIntegerTuple22._1()+integerIntegerTuple2._1(),integerIntegerTuple22._2()+integerIntegerTuple2._2());
}
}
);
reducBykey.collect().forEach(System.out::println);
}
}
:
(1214,(6,1))
(1213,(3,1))
(1215,(3,1))
(1212,(2,2))
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
Spark 팁: 컴퓨팅 집약적인 작업을 위해 병합 후 셔플 파티션 비활성화작은 입력에서 UDAF(사용자 정의 집계 함수) 내에서 컴퓨팅 집약적인 작업을 수행할 때 spark.sql.adaptive.coalescePartitions.enabled를 false로 설정합니다. Apache Sp...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.