Пишу в hadoop hdfs. Файл необходимо сжать с помощью lzo. Также файл будет добавлен в реальном времени.
Исходный файл - это файл gzip, которого нет в hadoop. Пакет обрабатывает этот файл gzip, а затем выполняет сжатие lzo и добавляет его в файл hadoop. Исключает ли это возможность использования map reduce?
Как мы можем этого добиться?
Заранее спасибо за помощь
Вы можете писать прямо в HDFS из пользовательского кода Java:
public class HdfsWrite extends Configured implements Tool {
public int run(String[] arg0) throws Exception {
// create am HDFS file system
FileSystem fs = FileSystem.get(getConf());
// create an output stream to write to a new file in hdfs
Path outputPath = new Path(
"/path/to/file/in/hdfs.default");
OutputStream outputStream = fs.create(outputPath);
// now wrap the output stream with a Zlib compression codec
CompressionCodecFactory codecFactory = new CompressionCodecFactory(getConf());
CompressionCodec codec = codecFactory.getCodec(outputPath);
CompressionOutputStream compressedOutput = codec.createOutputStream(outputStream);
// send content to file via compressed output stream using .write methods
// ..
// close out stream
compressedOutput.close();
return 0;
}
public static void main(String[] args) throws Exception {
ToolRunner.run(new HdfsWrite(), args);
}
}
Этот код работает для сжатия zlib - для сжатия LZO у вас уже есть библиотека Java, которая может выполнять сжатие за вас (например, библиотека hadoop-gpl-сжатия). Если вы установите указанную выше библиотеку, как описано, то все, что вам нужно сделать, это изменить расширение выходного пути на «.lzo_deflate», и все должно работать. Если вы хотите использовать другую библиотеку сжатия, вы можете пропустить блок кода CompressionCodecFactory и напрямую обернуть outputStream.
Что касается добавления к файлам - в зависимости от вашей версии hadoop это может не поддерживаться. Вам также необходимо подумать, поддерживает ли ваша библиотека сжатия объединенные файлы (например, GZip поддерживает, но при работе с этими типами в более ранних версиях Java / hadoop возникают некоторые проблемы). Если у вас есть версия hadoop, поддерживающая добавление, и ваша библиотека сжатия поддерживает это, измените вызов fs.create(outputPath) на fs.append(outputPath)
