Принцип write-once-read-many,на котором построена HDFS,не позволяет нескольким процессам/клиентам одновременно писать в один файл. А как насчет каталога? Могут несколько процессов одновременно создавать и писать в файлы в одном каталоге? Не смог найти в документации ответ на этот вопрос (может и плохо искал,не спорю). Могу,конечно,накидать тестовый пример параллельной записи для проверки,но вдруг есть готовый ответ?
1 ответ
Хм,раз никто не ответил,пришлось проверять практически. Да,параллельно писать в каталог можно. Нижеприведенный код корректно отрабатывает.
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;
}
}