스파크 RDD 기능

Question

내가 SparkContext 통해 CSV 파일을로드하려고 로딩 후 내가 CSV 파일의 필요한 열을 어떤 RDD 작업을 수행해야 할 파일. 나는 csv 파일을 읽고 JavaRDD에서 필수 열을 가져올 수있었습니다. 이제는 이러한 열에 대한 RDD 작업을 수행해야합니다.

이 내가 지금까지 시도한 것입니다 ..

JavaRDD<String> diskfile = sc.textFile("/Users/hadoop/Downloads/Data_1.csv"); 

    JavaRDD<Object> newRDD = diskfile.cache().map(lines -> Arrays.asList(new String[]{ 
      lines.split(",")[0], 
      lines.split(",")[1] 
     } 

     )); 
System.out.println(newRDD.collect()); 

newRDD.collect() 인쇄 0 번째 열 및 CSV 데이터의 1 열. 이제 newRDD에서 RDD 작업을 수행해야합니다.

미리 감사드립니다. 당신이 점화 RDD 변환을 시작 싶어 경우

Answer 1

당신은 다음 링크로 이동 할 수 있습니다

RDD-Basic-Examples

RDD API Examples

말씀이 자바 예를 카운트 flatMap, mapToPair, reduceByKey 변환 사용 :

JavaRDD<String> textFile = sc.textFile("hdfs://..."); 
JavaRDD<String> words = textFile.flatMap(new FlatMapFunction<String, String>() { 
    public Iterable<String> call(String s) { return Arrays.asList(s.split(" ")); } 
}); 
JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { 
    public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); } 
}); 
JavaPairRDD<String, Integer> counts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { 
    public Integer call(Integer a, Integer b) { return a + b; } 
}); 
counts.saveAsTextFile("hdfs://..."); 

을