MapReduce'da bir ara döngüsü aracılığıyla ara anahtarları yayar - MapReduce'u yanlış anlamış mıyım? Temel örnek dahil.

Question

Yani e-postaların bir koleksiyona sahip ve ne yapmak istediğinizi şöyle çıkış benzersiz üçüz (gönderici e-posta, alıcı e-posta, zaman damgası) için bunları kullanın geçerli:

user1@stackoverflow.com user2@stackoverflow.com 09/12/2009 16:45 
user1@stackoverflow.com user9@stackoverflow.com 09/12/2009 18:45 
user3@stackoverflow.com user4@stackoverflow.com 07/05/2008 12:29 

Yukarıdaki örnekte kullanıcı 1 tek gönderilen birden fazla alıcıya e-postayla (kullanıcı 2 ve kullanıcı 9). . Alıcıları kaydetmek için, bir Damgasıyla yanı sıra Sender ve Alıcı e-posta adreslerini tutacak bir veri yapısını EdgeWritable (WritableComparable) uygulayan oluşturulan

Benim mapper şöyle görünür:

private final EdgeWritable edge = new EdgeWritable(); // Data structure for triplets. 
private final NullWritable noval = NullWritable.get(); 

... 

@Override 
public void map(Text key, BytesWritable value, Context context) 
     throws IOException, InterruptedException { 

    byte[] bytes = value.getBytes(); 
    Scanner scanner = new Scanner(new ByteArrayInputStream(bytes), "UTF-8"); 
    String from = null; // Sender's Email address 
    ArrayList<String> recipients = new ArrayList<String>(); // List of recipients' Email addresses 
    long millis = -1; // Date 

    // Parse information from file 
    while(scanner.hasNext()) { 
     String line = scanner.nextLine(); 
     if (line.startsWith("From:")) { 
      from = procFrom(stripCommand(line, "From:")); // Get sender e-mail address. 
     } else if (line.startsWith("To:")) { 
      procRecipients(stripCommand(line, "To:"), recipients); // Populate recipients into a list. 
     } else if (line.startsWith("Date:")) { 
      millis = procDate(stripCommand(line, "Date:")); // Get timestamp. 

     if (line.equals("")) { // Empty line indicates the end of the header 
      break; 
     } 
    } 
    scanner.close(); 

    // Emit EdgeWritable as intermediate key containing Sender, Recipient and Timestamp. 
    if (from != null && recipients.size() > 0 && millis != -1) { 
     //EdgeWritable has 2 Text values (ew[0] and ew[1]) and a Timestamp. ew[0] is the sender, ew[1] is a recipient. 
     edge.set(0, from); // Set ew[0] 
     for(int i = 0; i < recipients.size(); i++) { 
      edge.set(1, recipients.get(i)); // Set edge from sender to each recipient i. 
      edge.setTS(millis); // Set date. 
      context.write(edge, noval); // Emit the edge as an intermediate key with a null value. 
     } 
    } 
} 

... 

Benim düşürücü basitçe tarihini biçimlendirir ve kenarları verir:

public void reduce(EdgeWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { 
    String date = MailReader.sdf.format(edge.getTS()); 
    out.set(edge.get(0) + " " + edge.get(1) + " " + date); // same edge from Mapper (an EdgeWritable). 
    context.write(noval, out); // same noval from Mapper (a NullWritable). 
} 

Ara anahtar olarak EdgeWritable'ı ve değer olarak (mapper'de) olarak NullWritable'ı kullanmak bir gereksinimdir, başka yöntemler kullanmasına izin verilmemelidir. Bu benim ilk Hadoop/MapReduce programım ve sadece doğru yönde ilerlediğimi bilmek istedim. Çok sayıda MapReduce örneğine çevrimiçi baktım ve yaptığım gibi bir döngüde anahtar/değer çiftlerini hiç görmedim. Burada bir tür hile kaçırıyormuşum gibi hissediyorum, ama bu şekilde bir döngü kullanmak düşünebileceğim tek yaklaşım.

Bu 'kötü' mi? Umarım bu açıktır, ancak daha fazla açıklama gerekirse lütfen bana bildirin.

Answer 1

Her bir kayıt için harita yöntemi çağrılır, böylece dizi listeniz her çağrı için yalnızca 1 kayıt alır. Dizi listenizi sınıf düzeyinde bildirin, böylece tüm kayıtların değerlerini saklayabilirsiniz. Sonra temizleme yönteminde harita içinde yazdığınız emit mantığını yapabilirsiniz. Bunu dene ve çalışırsa bana haber ver.