Changeset 1700:f6414830e65b

Timestamp:

07/14/08 09:46:14 (6 weeks ago)

Author:

hadoopsandholm@…

Branch:

default

Message:

demo backend updates

Location:

src/grid/hadoop/bin

Files:

• JobCompile.py (modified) (2 diffs)
• admin-server (modified) (2 diffs)
• th_stop-admin-server (modified) (1 diff)
• th_upload (modified) (1 diff)

src/grid/hadoop/bin/JobCompile.py

@@ -14 +14 @@
     mapper = cols[0]
     reducer = cols[1]
-    numreducers = cols[2]
+    nummaps = cols[2]
+    numreducers = cols[3]
+    if nummaps == 'max':
+        nummaps = nodes * 2
+    nummaps = long(nummaps)
     if numreducers == 'max':
         numreducers = nodes * 2
@@ -22 +26 @@
     outdir = inputdir + ".out%d" % job
     job += 1 
-    cmd += "hadoop dfs -rmr /%s; hadoop jar /home/guest/hadoop/build/contrib/streaming/hadoop-streaming.jar -input /%s/* -output /%s -mapper %s -cmdenv INDIR=%s -cmdenv OUTDIR=%s -file %s -file %s -reducer %s -jobconf mapred.reduce.tasks=%d -jobconf mapred.job.name='%s';" % (outdir, inputdir, outdir, mapper, inputdir, outdir, mapper, reducer, reducer, numreducers, mapper)
+    cmd += "hadoop dfs -rmr /%s; hadoop jar /home/guest/hadoop/build/contrib/streaming/hadoop-streaming.jar -input /%s/* -output /%s -mapper %s -cmdenv INDIR=%s -cmdenv OUTDIR=%s -file %s -file %s -reducer %s -jobconf mapred.map.tasks=%d -jobconf mapred.reduce.tasks=%d -jobconf mapred.job.name='%s';" % (outdir, inputdir, outdir, mapper, inputdir, outdir, mapper, reducer, reducer, nummaps, numreducers, mapper)
 print cmd[0:-1]

src/grid/hadoop/bin/admin-server

@@ -113 +113 @@
                 f.write(fileitem[0])
                 f.close()
-                stdout, stdin, stderr = popen3("tycoon_hadoop upload " + filepath)
+                stdout, stdin, stderr = popen3("tycoon_hadoop asynch upload " + filepath)
                 if stderr != None:
                     stderr_str = stderr.read().strip()
@@ -277 +277 @@
                     return
         self.text_headers()
-        self.wfile.write(stdout.read().strip())
+        if params[0] == 'asynch-get':
+            self.wfile.write(stdout.read().strip().replace('\n','<br>'))
+        else: 
+            self.wfile.write(stdout.read().strip())
 
     def authorize(self):

src/grid/hadoop/bin/th_stop-admin-server

@@ -1 +1 @@
 #! /bin/bash
-WEBPORT=`cat ${HADOOP_ROOT}/hadoop.conf | grep admin_port | awk '{print $2}'`
-PID=`cat ${HADOOP_ROOT}/.adminport`
+USER=`id -un`
+URLROOT=`cat ${HADOOP_ROOT}/hadoop.conf | grep url_root | awk '{print $2}'`
+PID=`cat ${URLROOT}/${USER}/.adminport`
 kill ${PID}

src/grid/hadoop/bin/th_upload

@@ -1 +1 @@
 #! /bin/bash
-STREAMINDIR=`tycoon_hadoop get-config stream_indir`
+FILENAME=$1
+STREAMINDIR=$2
+if [ "X$STREAMINDIR" == "X" ]; then
+  STREAMINDIR=`tycoon_hadoop get-config stream_indir`
+fi
 MASTER=`cat ${HADOOP_ROOT}/.master`
 ${SSHCMD} ${MASTER} "rm -rf /tmp/${STREAMINDIR}; mkdir /tmp/${STREAMINDIR}"
-${SCPCMD} ${1} ${MASTER}:/tmp/${STREAMINDIR}/upload.tar.gz
-${SSHCMD} ${MASTER} "tar -C /tmp/${STREAMINDIR} -zxvf /tmp/${STREAMINDIR}/upload.tar.gz; rm /tmp/${STREAMINDIR}/upload.tar.gz; hadoop dfs -mkdir /${STREAMINDIR}; hadoop dfs -copyFromLocal /tmp/${STREAMINDIR}/* /${STREAMINDIR}/; rm -rf /tmp/${STREAMINDIR};"
-
+TARBALL=`echo ${FILENAME} | grep 'tar.gz'`
+if [ "X$TARBALL" == "X" ]; then
+  LOCALPART=`echo $FILENAME | sed 's/.*\///'`
+  ${SCPCMD} ${FILENAME} ${MASTER}:/tmp/${STREAMINDIR}/${LOCALPART}
+  ${SSHCMD} ${MASTER} "hadoop dfs -mkdir /${STREAMINDIR}; hadoop dfs -copyFromLocal /tmp/${STREAMINDIR}/${LOCALPART} /${STREAMINDIR}/; rm -rf /tmp/${STREAMINDIR};"
+else
+  ${SCPCMD} ${FILENAME} ${MASTER}:/tmp/${STREAMINDIR}/upload.tar.gz
+  ${SSHCMD} ${MASTER} "tar -C /tmp/${STREAMINDIR} -zxvf /tmp/${STREAMINDIR}/upload.tar.gz; rm /tmp/${STREAMINDIR}/upload.tar.gz; hadoop dfs -mkdir /${STREAMINDIR}; hadoop dfs -copyFromLocal /tmp/${STREAMINDIR}/* /${STREAMINDIR}/; rm -rf /tmp/${STREAMINDIR};"
+fi