- Improving the core.

- Adding a primary-key from dataset file.

examples/invert-index/task.php

@@ -9,16 +9,16 @@ final class InvertIndex extends Job
 {
     function __config()
     {
-        $this->setInput("noticias/*");
-        $this->setOutput("index");
+        $this->setInput("news.txt");
+        $this->setOutput("invert-index");
         $this->setReducers(1);
     }
 
     function map($key, &$value)
     {
         $words = array();
         $value = strtolower($value);
-        foreach (preg_split("/[^a-z]/i", $value) as $word) {
+        foreach (preg_split("/[^a-záéíóúüñ]/i", $value) as $word) {
             $words[$word] = 1;
         }
         foreach ($words as $word=>$id) {

examples/k-means/sample.php

@@ -10,6 +10,8 @@
 
 hadoop::setHome("/home/crodas/hadoop/hadoop-0.18.3");
 
+exit();
+
 $hadoop = new Hadoop;
 
 /* create an invert index for fast computation */

examples/pk/index.php

@@ -0,0 +1,16 @@
+<?php
+include("../../src/hadoop.php");
+
+include("task.php");
+
+
+hadoop::setHome("/home/crodas/hadoop/hadoop-0.18.3");
+
+/* run the task */
+try {
+    new Index();
+} catch (Exception $e) {
+    print $e->getMessage()."\n";
+}
+
+?>

examples/pk/task.php

@@ -0,0 +1,33 @@
+<?php
+
+/**
+ *  Invert Index
+ *
+ *  This class contains an Invert Index.
+ */
+final class Index extends Job
+{
+    function __config()
+    {
+        $this->setInput("news.txt");
+        $this->setOutput("index");
+        $this->setOption(HAS_NO_REDUCER);
+        $this->setMappers(1);
+    }
+
+    function map($key, &$value)
+    {
+        static $i=0;
+        if (trim($key) == "") {
+            throw new Exception("Invalid index $key $value");
+        }
+        $this->Emit(++$i, $key);
+    }
+
+    function reduce($key, &$values)
+    {
+    }
+
+}
+
+?>

src/hadoop.php

@@ -6,6 +6,8 @@
 ini_set("memory_limit", "200M");
 
 
+define("HAS_NO_REDUCER",    2);
+
 /**
  *  Hadoop Class.
  *
@@ -14,7 +16,7 @@
  */
 abstract class Hadoop
 {
-    private static $_path;
+    private static $_path = false;
     private static $_injob = false;
     private static $_fncWatch = false;
     private $_ipath = false;
@@ -23,6 +25,8 @@ abstract class Hadoop
     private $_tmp;
     private $_reduce;
     private $_map = false;
+    private $_nomap = false;
+    private $_noreduce = false;
 
     final function __construct()
     {
@@ -33,10 +37,20 @@ final function __construct()
         if ($this->_ipath === false || $this->_opath === false) {
             throw new Exception("No input or output path configured");
         }
+        if (!self::$_path) {
+            throw new Exception("No Hadoop-home");
+        }
         $this->_tmp = array(tempnam("/tmp", "map"), tempnam("/tmp", "red"));
         $this->_execTask();
     }
 
+    final protected function setOption($options)
+    {
+        if ($options & HAS_NO_REDUCER) {
+            $this->_noreduce = true;
+        }
+    }
+
     final static function import($file)
     {
         include(dirname(__FILE__)."/$file");
@@ -156,6 +170,7 @@ private function _buildTempFiles()
         /* save the map */
         $map = str_replace("/*name*/", $info->getName(), $map);
         $map = str_replace("/*include*/", $includ, $map);
+        $map = str_replace("/*hadoop-home*/", self::$_path, $map);
         if (array_search($info->getFileName(), get_included_files()) === 0) {
             $map = str_replace("/*class*/", $code, $map);
         }
@@ -165,6 +180,7 @@ private function _buildTempFiles()
         /* save the reduce */
         $reduce = str_replace("/*name*/", $info->getName(), $reduce);
         $reduce = str_replace("/*include*/", $includ, $reduce);
+        $reduce = str_replace("/*hadoop-home*/", self::$_path, $reduce);
         if (array_search($info->getFileName(), get_included_files()) === 0) {
             $reduce = str_replace("/*class*/", $code, $reduce);
         }
@@ -184,13 +200,17 @@ private function _getCmd()
         $opath   = $this->_opath;
         $path    = self::$_path;
 
-        $cmd = sprintf("%sbin/hadoop jar %s -input %s -output %s -mapper %s -reducer %s  -file %s -file %s -file %s", 
+        $cmd = sprintf("%sbin/hadoop jar %s -input %s -output %s -file %s -file %s -file %s", 
                 $path, $path.$jarpath, $ipath, $opath, 
-                basename($this->_getFileName("map")), 
-                basename($this->_getFileName("reduce")),
                 $this->_getFileName("map"), $this->_getFileName("reduce"),__FILE__
             );
 
+        $cmd .= " -mapper ".basename($this->_getFileName("map"));
+
+        if (!$this->_noreduce) {
+            $cmd .= " -reducer ".basename($this->_getFileName("reducer"));
+        }
+
         /* set number of mappers */
         if (is_int($this->_map)) {
             $cmd .= " -jobconf mapred.map.tasks=".$this->_map;

src/job.php

@@ -40,15 +40,16 @@ final function runMap()
     {
         $this->__init();
         while (($line = fgets(STDIN)) !== false) {
-            $line = substr($line, 0, strlen($line)-1);
+            $line = trim($line);
             if (strlen($line) == 0) {
                 continue;
             }
             $input = $this->mapParser($line);
-            if (count($input) == 1) {
-                $input[1] = $input[0];
-                $input[0] = null;
-            } 
+
+            if (trim($input[0]) == "") {
+                throw new Exception($line);
+            }
+
             $this->unserialize($input[1]);
             $this->map($input[0], $input[1]);
         }

src/map.php

@@ -5,6 +5,7 @@
 /*class*/
 
 hadoop::initMapper();
+hadoop::setHome("/*hadoop-home*/");
 
 $map = new /*name*/;
 $map->RunMap();

src/reduce.php

@@ -4,6 +4,7 @@
 
 /*class*/
 hadoop::initReducer();
+hadoop::setHome("/*hadoop-home*/");
 
 $map = new /*name*/;
 $map->RunReduce();