Czy można równolegle dołączać do pliku HDFS wielu klientów?

Question

Zasadniczo całe pytanie znajduje się w tytule. Zastanawiam się, czy możliwe jest dołączanie do pliku znajdującego się na HDFS z wielu komputerów jednocześnie? Coś jak przechowywanie strumienia zdarzeń stale wytwarzanych przez wiele procesów. Zamówienie nie jest ważne.

Przypominam, że przesłuchanie w jednej z prezentacji technicznych Google, że GFS obsługuje takie funkcje dołączania, ale próbuje niektórych ograniczonych testów z HDFS (albo z regularnym dopiskiem do pliku() lub z SequenceFile) nie wydaje się działać.

Dzięki,

Answer 1

Nie sądzę, że jest to możliwe w HDFS. Nawet jeśli nie zależy Ci na kolejności rekordów, zależy ci na kolejności bajtów w pliku. Nie chcesz, aby pisarz A napisał częściowy zapis, który następnie zostanie uszkodzony przez pisarza B. Jest to trudny problem, który HDFS rozwiązuje samodzielnie, a więc nie.

Utwórz plik na jednego pisarza. Przekaż wszystkie pliki do dowolnego pracownika MapReduce, który musi odczytać te dane. Jest to znacznie prostsze i pasuje do projektu HDFS i Hadoop. Jeśli kod inny niż MapReduce musi odczytać te dane jako jeden strumień, należy kolejno przesyłać każdy plik lub napisać bardzo szybkie zadanie MapReduce w celu skonsolidowania plików.

Answer 2

Google powinien mieć trochę bardziej zaawansowany system plików, który obsługuje wiele bardziej swobodny dostęp lepiej niż HDFS. Korzystają z BigTable znacznie częściej i obszerniej, co wymaga szybszej modyfikacji bloków i współbieżnego odczytu/zapisu bloku. Ale faktycznie możesz zaimplementować coś podobnego. Ostatnio zrobiłem to, pisząc webcrawlera.

Zasadniczo nie można sparaliżować IO. Więc musisz użyć kolejki i sekwencyjnie dołączyć do pliku sekwencji.

private final ConcurrentLinkedQueue<FetchResult> queue = new ConcurrentLinkedQueue<FetchResult>(); 
private final Configuration conf = new Configuration(); 
private SequenceFile.Writer writer = null; 
public boolean running = true; 

public FetchResultPersister() throws IOException { 
    FileSystem fs = FileSystem.get(conf); 
    Path out = new Path("files/crawl/result.seq"); 
    fs.delete(out, true); 
    writer = new SequenceFile.Writer(fs, conf, out, Text.class, Text.class); 
} 

public final void add(final FetchResult result) { 
    queue.offer(result); 
} 

@Override 
public final void run() { 
    long retrieved = 0L; 
    while (running) { 
     final FetchResult poll = queue.poll(); 
     if (poll != null) { 
      try { 
       writer.append(new Text(poll.url), asText(poll.outlinks)); 
       retrieved++; 
       if (retrieved % 100 == 0) { 
        System.out 
          .println("Retrieved " + retrieved + " sites!"); 
       } 
      } catch (IOException e) { 
       e.printStackTrace(); 
      } 
     } else { 
      try { 
       Thread.sleep(10000); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 
    } 
    // close etc omitted 
} 

Głównym założeniem jest to, że dysk IO nie blokuje obliczeń.

Zasadniczo używasz tylko ConcurrentLinkedQueue, który jest zsynchronizowany i dodajesz wyniki z różnych wątków. Jak widać, jest to również uruchomione w wątku, polling dla nowych wyników do zapisu w pliku sekwencji.

Jestem pewien, że GFS obsługuje te rzeczy natywnie, HDFS nie (w tym momencie) nie.

Answer 3

tylko FYI, prawdopodobnie Byłoby pełni obsługiwane w Hadoop 2.6.x, acorrding do elementu JIRA na oficjalnej stronie: https://issues.apache.org/jira/browse/HDFS-7203