Hadoop에 두 개의 다른 데이터 세트를 동시에 읽으려는 제안이 있으십니까?

Question

친애하는 hadooper : 나는 hadoop에 새로 왔으며 최근에 알고리즘을 구현하려고합니다.

이 알고리즘은 2 쌍의 노래마다 서로 다른 등급을 나타내는 행렬을 계산해야합니다. 나는 이미 이것을했고 출력은 HDFS에 저장 한 600000 * 600000 스파 스 매트릭스입니다. 이 데이터 세트를 A (크기 = 160G)라고합시다.

이제 특정 노래에 대한 등급을 예측하기 위해 사용자의 프로필을 읽어야합니다. 그래서 사용자의 프로필 (5G 크기)을 먼저 읽고이 데이터 세트 B를 호출 한 다음 데이터 세트 A를 계산해야합니다.

하지만 이제는 두 데이터 세트를 읽는 방법을 모르겠습니다. 단일 hadoop 프로그램. 또는 데이터 세트 B를 RAM으로 읽어 들여 계산을 할 수 있습니까? (HDFS가 배포 시스템이기 때문에 데이터 세트 B를 단일 컴퓨터의 메모리로 읽을 수 없기 때문에 그렇게 할 수 없습니다).

제안 사항?

Answer 1

Haddop을 사용하면 다른 폴더에 대해 다른지도 입력 형식을 사용할 수 있습니다. 따라서 여러 데이터 소스에서 읽은 다음지도 함수에서 특정 유형으로 캐스팅 할 수 있습니다. 즉, 다른 (String, SongSongRaiting)에 (String, User)가 있고 Map signature가 (String, Object) 인 경우입니다. 두 번째 단계는 선택 recomendation 알고리즘입니다, 그래서 어떤 방법으로 agregator calcualte recomendation에 대한 최소한의 정보를 가지게됩니다 데이터를 가입하십시오.

Answer 2

지도 함수 두 개를 사용할 수 있습니다. 각지도 함수 다른 처리를 구현하려는 경우 하나의 데이터 집합을 처리 할 수 ​​있습니다. 당신은 직장 conf와지도를 등록해야합니다. 예 :

  public static class FullOuterJoinStdDetMapper extends MapReduceBase implements Mapper <LongWritable ,Text ,Text, Text> 
    { 
      private String person_name, book_title,file_tag="person_book#"; 
      private String emit_value = new String(); 
      //emit_value = ""; 
      public void map(LongWritable key, Text values, OutputCollector<Text,Text>output, Reporter reporter) 
        throws IOException 
      { 
        String line = values.toString(); 
        try 
        { 
          String[] person_detail = line.split(","); 
          person_name = person_detail[0].trim(); 
          book_title = person_detail[1].trim(); 
        } 
        catch (ArrayIndexOutOfBoundsException e) 
        { 
          person_name = "student name missing"; 
        } 
        emit_value = file_tag + person_name; 
        output.collect(new Text(book_title), new Text(emit_value)); 
      } 

    } 


     public static class FullOuterJoinResultDetMapper extends MapReduceBase implements Mapper <LongWritable ,Text ,Text, Text> 
    { 
      private String author_name, book_title,file_tag="auth_book#"; 
      private String emit_value = new String(); 

// emit_value = ""; 공개 무효화 맵 (LongWritable 키, 텍스트 값, OutputCollectoroutput, 리포터 리포터) throw IOException { String line = values.toString(); try { String [] author_detail = line.split (","); author_name = author_detail [1] .trim(); book_title = author_detail [0] .trim(); } catch (ArrayIndexOutOfBoundsException e) { author_name = "시험에 출전하지 않았습니다."; }

      emit_value = file_tag + author_name;          
         output.collect(new Text(book_title), new Text(emit_value)); 
       } 

      } 


     public static void main(String args[]) 
        throws Exception 
    { 

      if(args.length !=3) 
        { 
          System.out.println("Input outpur file missing"); 
          System.exit(-1); 
        } 


      Configuration conf = new Configuration(); 
      String [] argum = new GenericOptionsParser(conf,args).getRemainingArgs(); 
      conf.set("mapred.textoutputformat.separator", ","); 
      JobConf mrjob = new JobConf(); 
      mrjob.setJobName("Inner_Join"); 
      mrjob.setJarByClass(FullOuterJoin.class); 

      MultipleInputs.addInputPath(mrjob,new Path(argum[0]),TextInputFormat.class,FullOuterJoinStdDetMapper.class); 
      MultipleInputs.addInputPath(mrjob,new Path(argum[1]),TextInputFormat.class,FullOuterJoinResultDetMapper.class); 

      FileOutputFormat.setOutputPath(mrjob,new Path(args[2])); 
      mrjob.setReducerClass(FullOuterJoinReducer.class); 

      mrjob.setOutputKeyClass(Text.class); 
      mrjob.setOutputValueClass(Text.class); 

      JobClient.runJob(mrjob); 
    }