Слияние небольших файлов в hadoop

Question

У меня есть каталог (Final Dir) в HDFS, в котором некоторые файлы (например, 10 мб) загружаются каждую минуту. Через некоторое время я хочу объединить все мелкие файлы в большой файл (например: 100 мб). Но пользователь постоянно толкает файлы в Final Dir. это непрерывный процесс.

Так что в первый раз мне нужно совместить первые 10 файлов с большим файлом (например: large.txt) и сохранить файл в Finaldir.

Теперь у меня вопрос, как я получу следующие 10 файлов, исключая первые 10 файлов?

может кто-нибудь, пожалуйста, помогите мне

Answer 1

Вот еще один альтернативный, это еще наследие подход указывал в своих @ Андрей комментарии, но с дополнительными шагами делает ваш входную папку в качестве буфера для приема небольших файлов, которые своевременно перенаправляют их в каталог tmp и объединяют их и возвращают результат обратно.

шаг 1: создать каталог TMP

hadoop fs -mkdir tmp 

шаг 2: переместить все маленькие файлы в каталог TMP в момент времени

hadoop fs -mv input/*.txt tmp 

шаг 3 -merge маленьких файлов с помощь Hadoop-потоковая банка

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.6.0.jar \ 
        -Dmapred.reduce.tasks=1 \ 
        -input "/user/abc/input" \ 
        -output "/user/abc/output" \ 
        -mapper cat \ 
        -reducer cat 

шаг 4- шаг выход к входной папке

hadoop fs -mv output/part-00000 input/large_file.txt 

шаг 5 - удалить выход

hadoop fs -rm -R output/ 

шаг 6 - удалить все файлы из TMP

hadoop fs -rm tmp/*.txt 

Создать скрипт из шага 2 до шага 6 и график его выполнения с регулярными интервалами для слияния небольших файлов с регулярными интервалами (может быть на каждую минуту, исходя из ваших потребностей)

шагов для планирования хрон для объединения маленьких файлов

шага 1: создать скрипт /дома/а/mergejob.ш с помощью описанных выше шагов (от 2 до 6)

Важное примечание: вам необходимо указать абсолютный путь Hadoop в сценарии следует понимать под хрон

#!/bin/bash 
/home/abc/hadoop-2.6.0/bin/hadoop fs -mv input/*.txt tmp 
wait 
/home/abc/hadoop-2.6.0/bin/hadoop jar /home/abc/hadoop-2.6.0/share/hadoop/tools/lib/hadoop-streaming-2.6.0.jar \ 
        -Dmapred.reduce.tasks=1 \ 
        -input "/user/abc/input" \ 
        -output "/user/abc/output" \ 
        -mapper cat \ 
        -reducer cat 
wait 
/home/abc/hadoop-2.6.0/bin/hadoop fs -mv output/part-00000 input/large_file.txt 
wait 
/home/abc/hadoop-2.6.0/bin/hadoop fs -rm -R output/ 
wait 
/home/abc/hadoop-2.6.0/bin/hadoop fs -rm tmp/*.txt 

шаг 2: график скрипт, используя хрон запускать каждую минуту с помощью выражения CRON

а) редактировать кронтаб путем выбора редактора

>crontab -e 

б) добавьте следующую строку в конце и выйти из редактора

* * * * * /bin/bash /home/abc/mergejob.sh > /dev/null 2>&1 

Работа слияния будет запланировано бежать за каждую минуту.

Надеюсь, это было полезно.

Answer 2

указал вам @ Андрей к решению, которое было соответствующих 6 лет назад, в периодическом ориентированном мире.
Но это 2016 год, у вас есть поток микро-пакетных данных и требуется неблокирующее решение.

Вот как бы я это сделать:

• создать внешнюю таблицу с 3-х разделов, отображенный на 3 каталогов например new_data, reorg и history
• подпитывать новые файлы в new_data
• осуществлять работу, чтобы запустить пакетное уплотнению, и запустить его периодически

Теперь партия уплотнительная логика:

1. убедитесь что ни один запрос SELECT не будет выполнен во время работы уплотнения, иначе он будет возвращать дубликаты
2. выберите все файлы, которые созревают для уплотнения (определить свою собственную критерии) и движения их из new_data каталога в reorg
3. слияния содержания всех эти reorg файлов в новый файл в history реже (не стесняйтесь GZIP его муха, Hive распознает .gz расширение)
4. падение файлы в reorg

Итак, это в основном старая история 2010 года, за исключением того, что существующий поток данных может продолжать сбрасывать новые файлы в new_data, а уплотнение безопасно работает в отдельных каталогах. И в случае сбоя работы уплотнения вы можете безопасно исследовать/очистить/возобновить уплотнение без ущерба для потока данных.

---

Кстати, я не большой поклонник решения 2010 года, основанного на работе «Hadoop Streaming» - с одной стороны, «потоковая передача» имеет совсем другое значение; с другой стороны, «потоки Hadoop» были полезны в старые времена, но теперь вышли из радара; на захватной руке [*] вы можете сделать это довольно просто с помощью запроса на улей, например.

INSERT INTO TABLE blahblah PARTITION (stage='history') 
SELECT a, b, c, d 
FROM blahblah 
WHERE stage='reorg' 
; 

С парой SET some.property = somevalue до этого запроса, вы можете определить, что кодек сжатия будет применен на файл результата (ов), сколько файлов (ы) вы хотите (или более точно, насколько велика вы хотите файлы быть - улей будет работать слияние соответственно), и т.д.

Посмотрите в https://cwiki.apache.org/confluence/display/Hive/Configuration+Properties под hive.merge.mapfiles и hive.merge.mapredfiles (или hive.merge.tezfiles, если вы используете Tez) и hive.merge.smallfiles.avgsize, а затем hive.exec.compress.output и mapreduce.output.fileoutputformat.compress.codec - плюс hive.hadoop.supports.splittable.combineinputformat уменьшить количество Карты, поскольку ваши входные файлы довольно малы.

[*] очень старый SF ссылка здесь :-)