Hendrik/xgboost
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update submit_job_hadoop.py

Browse Source
This commit is contained in:
Tianqi Chen 2014-12-05 09:14:44 -08:00
parent e05098cacb
commit e7a22792ac
1 changed files with 7 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
submit_job_hadoop.py
View File

@ -7,15 +7,7 @@ import sys
import os
import subprocess
sys.path.append('./src/')
from rabit_master import Master
from threading import Thread
def hadoop_streaming(nslaves, slave_args):
cmd = '%s jar %s -input %s -output %s -mapper \"%s %s\" -reducer /bin/cat stdin %d %d stdout' % (args.hadoop_binary, args.hadoop_streaming_jar, args.input, args.output, args.mapper, ' '.join(slave_args), args.nclusters, args.iterations)
print cmd
subprocess.check_call(cmd, shell = True)
import rabit_master as master
parser = argparse.ArgumentParser(description='Hadoop Streaming submission script')
parser.add_argument('-s', '--nslaves', required=True, type=int)
@ -29,11 +21,9 @@ parser.add_argument('-k', '--nclusters', required=True, type=int)
parser.add_argument('-itr', '--iterations', required=True, type=int)
args = parser.parse_args()
master = Master()
# this is availabe after triggered the hadoop streaming job, not sure how to do it
# os.environ["mapred_job_id"]
slave_args = ['master_uri=%s' % 'TODO', 'master_port=%s' % 'TODO']
submit_thread = Thread(target = hadoop_streaming, args = slave_args)
submit_thread.start()
master.accept_slaves(args.nslaves)
submit_thread.join()
def hadoop_streaming(nslaves, slave_args):
cmd = '%s jar %s -input %s -output %s -mapper \"%s %s\" -reducer /bin/cat stdin %d %d stdout' % (args.hadoop_binary, args.hadoop_streaming_jar, args.input, args.output, args.mapper, ' '.join(slave_args), args.nclusters, args.iterations)
print cmd
subprocess.check_call(cmd, shell = True)
master.submit(args.nslaves, [], fun_submit= hadoop_streaming)