0

Принцип write-once-read-many,на котором построена HDFS,не позволяет нескольким процессам/клиентам одновременно писать в один файл. А как насчет каталога? Могут несколько процессов одновременно создавать и писать в файлы в одном каталоге? Не смог найти в документации ответ на этот вопрос (может и плохо искал,не спорю). Могу,конечно,накидать тестовый пример параллельной записи для проверки,но вдруг есть готовый ответ?

1 ответ 1

0

Хм,раз никто не ответил,пришлось проверять практически. Да,параллельно писать в каталог можно. Нижеприведенный код корректно отрабатывает.

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ParallelDirectoryWrite extends Configured implements Tool{

    public static void main(String[] args) throws Exception {
         int res = ToolRunner.run(new Configuration(), new ParallelDirectoryWrite(), args);
         System.exit(res);
    }

    @Override
    public int run(String[] args){


     try {

        CountDownLatch latch = new CountDownLatch(1);
        Configuration conf = this.getConf();

        Thread th1 = new Thread(new FileWriteThread(conf,new Path("/ptest/fileA"),512,latch));
        Thread th2 = new Thread(new FileWriteThread(conf,new Path("/ptest/fileB"),512,latch));

        th1.start();
        th2.start();

        System.out.println("Launching threads...");
        latch.countDown();

        th1.join();
        th2.join();

        System.out.println("Main thread finished");

        return 0;

      }catch (Exception e) {
            e.printStackTrace();
            return 1;
      }

    }

}

class FileWriteThread implements Runnable {

    private Configuration conf;
    private Path file;
    private int size;
    private CountDownLatch latch;
    private final int MEGABYTE=1_048_576;

    public void run() {

        try {

            latch.await();
            System.out.println("Writing "+file.getName()+" started at "+now());
            FileSystem fs  = FileSystem.get(conf);
            if(fs.exists(file))
                fs.delete(file, false);
            FSDataOutputStream out = fs.create(file);
            int total = size*MEGABYTE;
            for(int i=1;i<=total;i++) {
                out.writeByte((int)'y');
                if(i % MEGABYTE == 0)
                    out.hflush();
            }
            out.close();
            System.out.println("Writing "+file.getName()+" finished at "+now());

        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private String now() {
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());
    }

    public FileWriteThread(Configuration conf, Path file, int size, CountDownLatch latch) {
        this.conf = conf;
        this.file = file;
        this.size = size;
        this.latch = latch;
    }

}

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.