hadoop - Concatenate S3 files to read in EMR -

i have s3 bucket log files want concatenate, use input emr job. log files in paths like: bucket-name/[date]/product/out/[hour]/[minute-based-file]. i'd take minute logs in hour directories in date directories, , concatenate them 1 file. want use file input emr job. original log files need preserved, , new combined log file written different s3 bucket.

i tried using hadoop fs -getmerge on emr master node via ssh, got error:

this file system object (file:///) not support access request path 's3://target-bucket-name/merged.log'

the source s3 bucket has other files in it, don't want include all of files. wildcard match looks this: s3n://bucket-name/*/product/out/*/log.*.

the purpose around problem of having tens/hundreds of thousands of small (10k-3mb) input files emr, , instead give 1 large file can split more efficiently.

i ended writing script wraps hadoop filesystem commands this.

#!/usr/bin/env ruby  require 'date'  # merge minute-based log files daily log files # usage: run on emr master (e.g. ssh master `ruby ~/merge-historical-logs.rb [from [to]]`)  source_bucket_name      = 's3-logs-bucket' destination_bucket_name = 's3-merged-logs-bucket'  # optional date inputs min_date = if argv[0]   min_date_args = argv[0].split('-').map {|item| item.to_i}   date.new(*min_date_args) else   date.new [2012, 9, 1] end  max_date = if argv[1]   max_date_args = argv[1].split('-').map {|item| item.to_i}   date.new(*max_date_args) else   date.today end  # setup directories hdfs_logs_dir = '/mnt/tmp/logs' local_tmp_dir = './_tmp_merges'  puts "cleaning filesystem" system "hadoop fs -rmr #{hdfs_logs_dir}" system "rm -rf #{local_tmp_dir}*"  puts "making hdfs directories" system "hadoop fs -mkdir #{hdfs_logs_dir}"  # progress backwards, max min date = max_date while date >= min_date   # format date pieces   year  = date.year   month = "%02d" % date.month   day   = "%02d" % date.day    # make directory in hdfs store day's hourly logs   today_hours_dir = "#{hdfs_logs_dir}/#{year}-#{month}-#{day}"   puts "making today's hourly directory"   system "hadoop fs -mkdir #{today_hours_dir}"    # break day's hours few chunks   # seems avoid problems when run lots of getmerge commands in parallel   [*(0..23)].each_slice(8).to_a.each |hour_chunk|     hour_chunk.each |_hour|       hour = "%02d" % _hour        # setup args merge minute logs hour logs       source_file = "s3://#{source_bucket_name}/#{year}-#{month}-#{day}/product/out/#{hour}/"       output_file = "#{local_tmp_dir}/#{hour}.log"        # launch each hour's getmerge in background       full_command = "hadoop fs -getmerge #{source_file} #{output_file}"       puts "forking: #{full_command}"       fork { system full_command }     end      # wait batch of germerge's finish     process.waitall   end    # delete local temp files hadoop created   puts "removing temp files"   system "rm #{local_tmp_dir}/.*.crc"    # move local hourly logs hdfs free local space   puts "moving local logs hdfs"   system "hadoop fs -put #{local_tmp_dir}/* #{today_hours_dir}"    puts "removing local logs"   system "rm -rf #{local_tmp_dir}"    # merge day's hourly logs single daily log file   daily_log_file_name = "#{year}-#{month}-#{day}.log"   daily_log_file_path = "#{local_tmp_dir}_day/#{daily_log_file_name}"   puts "merging hourly logs daily log"   system "hadoop fs -getmerge #{today_hours_dir}/ #{daily_log_file_path}"    # write daily log file s3 bucket   puts "writing daily log s3"   system "hadoop fs -put #{daily_log_file_path} s3://#{destination_bucket_dir}/daily-merged-logs/#{daily_log_file_name}"    # remove daily log locally   puts "removing local daily logs"   system "rm -rf #{local_tmp_dir}_day"    # remove hourly logs hdfs   puts "removing hdfs hourly logs"   system "hadoop fs -rmr #{today_hours_dir}"    # go in time   date -= 1 end