CVE-2022-4641: Insecure Temporary File (CWE-377)
value: A vulnerability was found in pig-vector and classified as problematic. Affected by this issue is the function LogisticRegression of the file src/main/java/org/apache/mahout/pig/LogisticRegression.java. The manipulation leads to insecure temporary file. The attack needs to be approached locally. The name of the patch is 1e7bd9fab5401a2df18d2eabd802adcf0dcf1f15. It is recommended to apply a patch to fix this issue. The identifier of this vulnerability is VDB-216500.
CWE description: Creating and using insecure temporary files can leave application and system data vulnerable to attack.
CVSS v3 base score: 5.5
Published: 2022-12-21T22:15Z
Repository: https://github.com/tdunning/pig-vector
Language: Java
Files changed: .gitignore, README.md, pom.xml, src/main/antlr3/org/apache/mahout/pig/encoders/Formula.g, src/main/antlr3/org/apache/mahout/pig/encoders/Schema.g, src/main/java/org/apache/mahout/pig/BadClassifierSpecException.java, src/main/java/org/apache/mahout/pig/Classifier.java, src/main/java/org/apache/mahout/pig/ImpossibleStateError.java, src/main/java/org/apache/mahout/pig/InvalidOutputSchema.java, src/main/java/org/apache/mahout/pig/LogisticRegression.java, src/main/java/org/apache/mahout/pig/LogisticRegressionEval.java, src/main/java/org/apache/mahout/pig/MessageLoader.java, src/main/java/org/apache/mahout/pig/PigModelStorage.java, src/main/java/org/apache/mahout/pig/PigVector.java, src/main/java/org/apache/mahout/pig/encoders/ArgumentEncoder.java, src/main/java/org/apache/mahout/pig/encoders/DirectVector.java, src/main/java/org/apache/mahout/pig/encoders/EncodeVector.java, src/main/java/org/apache/mahout/pig/encoders/EncodingSpec.java, src/main/java/org/apache/mahout/pig/encoders/Formula.java, src/main/java/org/apache/mahout/pig/encoders/FormulaParseException.java, src/main/java/org/apache/mahout/pig/encoders/Schema.java, src/main/java/org/apache/mahout/pig/encoders/SchemaParseException.java, src/test/java/org/apache/mahout/pig/EncodeVectorTest.java, src/test/java/org/apache/mahout/pig/EncodingSpecTest.java, src/test/java/org/apache/mahout/pig/LogisticRegressionTest.java, src/test/java/org/apache/mahout/pig/SchemaTest.java, src/test/java/org/apache/mahout/pig/encoders/DirectVectorTest.java, src/test/pig/encoderTest.pig
Commit date: 2022-11-19 22:53:54 -0800
Commit message: vuln-fix: Temporary File Information Disclosure



This fixes temporary file information disclosure vulnerability due to the use
of the vulnerable `File.createTempFile()` method. The vulnerability is fixed by
using the `Files.createTempFile()` method which sets the correct posix permissions.

Weakness: CWE-377: Insecure Temporary File
Severity: Medium
CVSSS: 5.5
Detection: CodeQL & OpenRewrite (https://public.moderne.io/recipes/org.openrewrite.java.security.SecureTempFileCreation)

Reported-by: Jonathan Leitschuh <Jonathan.Leitschuh@gmail.com>
Signed-off-by: Jonathan Leitschuh <Jonathan.Leitschuh@gmail.com>

Bug-tracker: https://github.com/JLLeitschuh/security-research/issues/18


Co-authored-by: Moderne <team@moderne.io>
Diff: diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..098ad5e
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,3 @@
+.idea/*
+target/*
+*.iml
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..2089c5c
--- /dev/null
+++ b/README.md
@@ -0,0 +1,126 @@
+This code provides the ability to encode data in Pig using the hashed encoding capabilities of Mahout.  It also allows
+training of logistic regression models.
+
+To try this out, start by compiling and installing mahout in your local repository:
+
+    cd ~/Apache
+    git clone https://github.com/apache/mahout.git
+    cd mahout
+    mvn install -DskipTests
+
+Then do the same with pig-vector
+
+    cd ~/Apache
+    git clone git@github.com:tdunning/pig-vector.git
+    cd pig-vector
+    mvn package
+
+Now you have a comprehensive jar that includes pig-vector and Mahout.  This jar is big (24M) but it has a lot of stuff on its mind.
+
+Now you can write a program in pig that uses Mahout vectors.
+
+To do this, start off by registering the pig-vector jar:
+
+    register 'target/pig-vector-1.0-jar-with-dependencies.jar';
+
+Then define an encoding function.  The arguments to the constructor are 
+
+- the dimension of the resulting vectors (100,000 to 1,000,000 is a more common range than the 10 showed here)
+
+- a description of the variables to be included in the encoding.    The interactions currently don't work, but all the rest does.
+
+- the schema of the tuples that pig will pass together with their data types.
+
+There is a lot of flexibility here in terms of tokenizing text, reusing fields with different encodings and interaction variables.
+
+    define EncodeVector org.apache.mahout.pig.encoders.EncodeVector('10','x+y+1', 'x:numeric, y:numeric, z:numeric');
+
+The other important types of data are word (for categorical data) and text.  If you just say "text" is the type, then
+whitespace is used for tokenization.  You can also add a Lucene 3.1 tokenizer in parentheses if you want something fancier
+in the way of tokenization.  The encoder treats word fields as strings with no processing whatsoever.
+
+The formula is not a normal arithmetic formula, but is more akin to an R formula as used in regression.  In this style,
+you list the variables you want to include, joined by +'s.  You can add 1 if you want a constant terms which is very important
+for most generalized linear classifiers.  You can specify interactions using * or you can add all interactions of a group
+of terms by using exponentiation.  That allows something like (x+y+z)^3 which includes all first, second and third order
+variables.  The syntax doesn't yet allow the ':' operator of R and when it comes time to encoding no interactions are implemented
+yet, regardless of syntax support.
+
+Then we just need data to encode.  Here I read an output file that has some silly data in it.
+
+    a = load '/Users/tdunning/Downloads/NNBench.csv' using PigStorage(',') as (x1:int, x2:int, x3:int);
+
+and here is code that encodes all of the data (that we care about):
+
+    b = foreach a generate 1 as key, EncodeVector(*) as v;
+
+Note that the schema for the tuple has fields x1, x2, x3 while the schema for the encoding has x, y, z.
+Since the schema of the tuple is not visible to the UDF, we can't always avoid this kind of mismatch except by careful
+coding of the pig code.  Note also how the encoding of the vector (defined by the formula "x+y+1") includes fields x1
+and x2 and a bias term but excludes x3.
+
+You can look at the output
+
+    dump b;         
+
+The output is in binary form so you can't see it, but it has two non-zero fields that encode x1, two that
+encode x2 and a single field that encodes the bias. With only 10 dimensional output vectors, some of these may
+overlap and cause a larger value.
+
+Or you can write the vector out in a sequence file by using the elephant-bird package from twitter which is
+included in pig-vector:
+
+    store b into 'vectors.dat' using com.twitter.elephantbird.pig.store.SequenceFileStorage (
+       '-c com.twitter.elephantbird.pig.util.IntWritableConverter',
+       '-c com.twitter.elephantbird.pig.util.GenericWritableConverter -t org.apache.mahout.math.VectorWritable'
+    );
+
+This data isn't very interesting because the key for every vector is the constant value 1, but you can see the
+pattern.  Note how the IntWritableConverter is used to specify the type of the key and the GenericWritableConverter
+is used to specify how to write the vector value.  The -t option is required so that we can specify what the
+actual type of is for the data we want to write.
+
+Vectors encoded using DirectVector or EncodeVector are passed around in pig as bytearray's.  If they are passed
+to a learning algorithm, they have to be converted into real vector objects in order to be passed to the learning
+algorithm.  When writing them out to a file, we can just use the bytes as is with the help of the
+GenericWritableConverter class.
+
+To train a model, we use the org.apache.mahout.pig.LogisticRegression UDF.  The constructor for this function
+requires a set of options to specify how it should run.  Here is a sample:
+
+    define train org.apache.mahout.pig.LogisticRegression('iterations=5, inMemory=true, features=100000, categories=alt.atheism comp.sys.mac.hardware rec.motorcycles sci.electronics talk.politics.guns comp.graphics comp.windows.x rec.sport.baseball sci.med talk.politics.mideast comp.os.ms-windows.misc misc.forsale rec.sport.hockey sci.space talk.politics.misc comp.sys.ibm.pc.hardware rec.autos sci.crypt soc.religion.christian talk.religion.misc');
+
+This very long line specifies the number of learning iterations (5), that the training vectors should be kept
+in memory and that the feature vectors are have 100,000 elements.  The categories option is a space delimited
+list of the categories and must match the strings that are passed into the learning algorithm as the target variable.
+
+Here is a complete training script:
+
+    register 'target/pig-vector-1.0-jar-with-dependencies.jar';
+    define encodeVector org.apache.mahout.pig.encoders.EncodeVector('100000', 'subject+body', 'group:word, article:numeric, subject:text, body:text');
+    define train org.apache.mahout.pig.LogisticRegression('iterations=5, inMemory=true, features=100000, categories=alt.atheism comp.sys.mac.hardware rec.motorcycles sci.electronics talk.politics.guns comp.graphics comp.windows.x rec.sport.baseball sci.med talk.politics.mideast comp.os.ms-windows.misc misc.forsale rec.sport.hockey sci.space talk.politics.misc comp.sys.ibm.pc.hardware rec.autos sci.crypt soc.religion.christian talk.religion.misc');
+
+    rm model.dat
+
+    /* read the data */
+    docs = load '20news-bydate-train/*/*' using org.apache.mahout.pig.MessageLoader()
+         as (newsgroup, id:int, subject, body);
+    /* encode as vectors, retain the target variable and the feature vector */
+    vectors = foreach docs generate newsgroup, encodeVector(*) as v;
+    describe vectors;
+
+    /* put the training data in a single bag.  We could train multiple models this way */
+    grouped = group vectors all;
+
+    /* train the actual model.  The key is bogus to satisfy the sequence vector format. */
+    model = foreach grouped generate 1 as key, train(vectors) as model;
+
+    /* the trained model is passed to use as a bytearray so we just pass it on out.  The classifier
+       class just contains the list of target valeus and the OnlineLogisticRegression object itself. */
+    store model into 'model.dat' using com.twitter.elephantbird.pig.store.SequenceFileStorage (
+       '-c com.twitter.elephantbird.pig.util.IntWritableConverter',
+       '-c com.twitter.elephantbird.pig.util.GenericWritableConverter -t org.apache.mahout.pig.Classifier'
+    );
+
+Send questions and suggestions to ted.dunning@gmail.com
+
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..c38dbe2
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,99 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2014 Ted Dunning
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>pig-vector</groupId>
+    <artifactId>pig-vector</artifactId>
+    <version>1.0</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.pig</groupId>
+            <artifactId>pig</artifactId>
+            <version>0.9.1</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.twitter.elephantbird</groupId>
+            <artifactId>elephant-bird-core</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.mahout</groupId>
+            <artifactId>mahout-core</artifactId>
+            <version>0.8</version>
+        </dependency>
+        <dependency>
+            <groupId>org.antlr</groupId>
+            <artifactId>antlr</artifactId>
+            <version>3.4</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.antlr</groupId>
+                <artifactId>antlr3-maven-plugin</artifactId>
+                <version>3.4</version>
+
+                <executions>
+                    <execution>
+                        <id>antlr-generate</id>
+                        <!-- this is used for inheritance merges -->
+                        <phase>generate-sources</phase>
+                        <!-- bind to the packaging phase -->
+                        <goals>
+                            <goal>antlr</goal>
+                        </goals>
+                    </execution>
+                </executions>
+
+            </plugin>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>2.3.2</version>
+                <configuration>
+                    <source>1.6</source>
+                    <target>1.6</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/src/main/antlr3/org/apache/mahout/pig/encoders/Formula.g b/src/main/antlr3/org/apache/mahout/pig/encoders/Formula.g
new file mode 100644
index 0000000..0037c2b
--- /dev/null
+++ b/src/main/antlr3/org/apache/mahout/pig/encoders/Formula.g
@@ -0,0 +1,52 @@
+grammar Formula;
+
+options {
+  output = AST;
+}
+
+@header {
+package org.apache.mahout.pig.encoders;
+}
+
+@lexer::header {
+package org.apache.mahout.pig.encoders;
+}
+
+@members {
+  public void reportError(RecognitionException e) {
+    if (e instanceof FailedPredicateException) {
+      throw new FormulaParseException(e, "Invalid bias");
+    } else {
+      throw new FormulaParseException(e, "Syntax error in schema: ");
+    }
+  }
+}
+
+expression returns [EncodingSpec r]: s=sumExpression EOF  {$r = $s.r;};
+
+sumExpression returns [EncodingSpec r]:
+    f1=factor {$r = $f1.r;}
+       (PLUS f2=factor {$r = EncodingSpec.add($r, $f2.r);})*
+    ;
+
+factor returns [EncodingSpec r]:
+    a1=argument {$r = $a1.r;} (TIMES a2=argument {$r = EncodingSpec.interact($r, $a2.r);})*  ;
+	
+argument returns [EncodingSpec r]:
+    LPAREN s1=sumExpression  RPAREN  {$r = $s1.r;}
+    (UP_ARROW n = INTEGER {$r = EncodingSpec.pow($s1.r, Integer.parseInt($n.getText()));})?
+    | ID {$r = EncodingSpec.atom($ID.text);}
+    | bias {$r = EncodingSpec.atom("1");};
+
+bias : i = INTEGER {"1".equals(i.getText())}?;
+
+PLUS:       '+';
+TIMES:      '*';
+LPAREN:     '(';
+RPAREN:     ')';
+UP_ARROW:   '^';
+INTEGER:	('0'..'9')+;
+ID:         ('a'..'z'|'A'..'Z'|'_') ('a'..'z'|'A'..'Z'|'0'..'9'|'_')*;
+DOLLAR:     '$';
+WS       :           (' '|'\t'|'\f'|'\n'|'\r')+{ $channel=HIDDEN; };
+
diff --git a/src/main/antlr3/org/apache/mahout/pig/encoders/Schema.g b/src/main/antlr3/org/apache/mahout/pig/encoders/Schema.g
new file mode 100644
index 0000000..b8be469
--- /dev/null
+++ b/src/main/antlr3/org/apache/mahout/pig/encoders/Schema.g
@@ -0,0 +1,76 @@
+grammar Schema;
+
+options {
+  output = AST;
+}
+
+@header {
+package org.apache.mahout.pig.encoders;
+
+import com.google.common.collect.Maps;
+import java.util.Map;
+import java.util.HashMap;
+}
+
+@lexer::header {
+package org.apache.mahout.pig.encoders;
+}
+
+@members {       
+       private int position = 0;
+       private Map<String, ArgumentEncoder> encoders = Maps.newHashMap();
+     
+       public Map<String, ArgumentEncoder> getEncoders() {
+          return encoders;
+       }
+
+       public void reportError(RecognitionException e) {
+         if (e instanceof FailedPredicateException) {
+           throw new SchemaParseException("Variable multiply defined", e);
+         } else {
+           throw new SchemaParseException("Syntax error in schema: ", e);
+         }
+       }
+
+}
+
+schema 	:	{position = 0;} variable (COMMA! variable)*;
+
+variable	:
+          ID^ COLON! t=type[$ID.text, position++]
+            {encoders.get($ID.text) == null}?
+            {encoders.put($ID.text, $t.encoder);}
+          | ID^ DOLLAR INTEGER COLON! t=type[$ID.text, Integer.parseInt($INTEGER.text)]
+            {encoders.get($ID.text) == null}?
+            {encoders.put($ID.text, $t.encoder);};
+
+type [String name, int position] returns [ArgumentEncoder encoder]:	
+          NUMERIC^ {$encoder = ArgumentEncoder.newNumericEncoder(position, $name);}
+          | WORD^ {$encoder = ArgumentEncoder.newWordEncoder(position, $name);}
+          | TEXT^ {$encoder = ArgumentEncoder.newTextEncoder(position, $name);}
+          | TEXT^ LPAREN! className RPAREN! {$encoder = ArgumentEncoder.newTextEncoder(position, $name, $className.text);}
+          ;
+          
+className	:	ID (DOT ID)*;
+
+NUMERIC	:	'numeric';
+WORD	:	'word';
+TEXT	:	'text';
+INTEGER:	('0'..'9')+;
+
+LPAREN	:	'(';
+RPAREN	:	')';
+COLON	:	':';
+ID  :	('a'..'z'|'A'..'Z'|'_') ('a'..'z'|'A'..'Z'|'0'..'9'|'_')*
+    ;
+DOT	:	'.';
+COMMA	:	',';
+DOLLAR	:	'$';
+
+WS  :   ( ' '
+        | '\t'
+        | '\r'
+        | '\n'
+        ) {$channel=HIDDEN;}
+    ;
+
diff --git a/src/main/java/org/apache/mahout/pig/BadClassifierSpecException.java b/src/main/java/org/apache/mahout/pig/BadClassifierSpecException.java
new file mode 100644
index 0000000..e9f22a5
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/BadClassifierSpecException.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+/**
+ * Signal an error in classifier construction.
+ */
+public class BadClassifierSpecException extends RuntimeException {
+    public BadClassifierSpecException(String msg, Throwable e) {
+        super(msg, e);
+    }
+
+    public BadClassifierSpecException(String msg) {
+        super(msg);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/Classifier.java b/src/main/java/org/apache/mahout/pig/Classifier.java
new file mode 100644
index 0000000..cafa9c1
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/Classifier.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import com.google.common.collect.Lists;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import org.apache.pig.impl.util.UDFContext;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Writable wrapper for a classifier and a list of target categories.
+ */
+public class Classifier implements Writable {
+    private List<String> categories;
+    private OnlineLogisticRegression model;
+
+    public Classifier() {
+        categories = Lists.newArrayList();
+    }
+
+    public Classifier(List<String> categories, OnlineLogisticRegression model) {
+        this.categories = Lists.newArrayList(categories);
+        this.model = model;
+    }
+
+    public Classifier(String file) {
+        try {
+            Configuration conf = UDFContext.getUDFContext().getJobConf();
+            final FSDataInputStream in = FileSystem.get(conf).open(new Path(file));
+            try {
+                this.readFields(in);
+            } finally {
+                in.close();
+            }
+        } catch (FileNotFoundException e) {
+            throw new BadClassifierSpecException("Can't find file to read model from: " + file, e);
+        } catch (IOException e) {
+            throw new BadClassifierSpecException("Error while reading model from: " + file, e);
+        }
+    }
+
+    public void write(DataOutput dataOutput) throws IOException {
+        dataOutput.writeInt(categories.size());
+        for (String category : categories) {
+            new Text(category).write(dataOutput);
+        }
+        PolymorphicWritable.write(dataOutput, model);
+    }
+
+    public void readFields(DataInput dataInput) throws IOException {
+        int n = dataInput.readInt();
+        for (int i = 0; i < n; i++) {
+            Text tmp = new Text();
+            tmp.readFields(dataInput);
+            categories.add(tmp.toString());
+        }
+        model = PolymorphicWritable.read(dataInput, OnlineLogisticRegression.class);
+    }
+
+    public List<String> getCategories() {
+        return categories;
+    }
+
+    public OnlineLogisticRegression getModel() {
+        return model;
+    }
+
+}
diff --git a/src/main/java/org/apache/mahout/pig/ImpossibleStateError.java b/src/main/java/org/apache/mahout/pig/ImpossibleStateError.java
new file mode 100644
index 0000000..0a80a33
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/ImpossibleStateError.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+/**
+ * Indicates an "impossible" condition has been detected that prevents further progress.
+ */
+public class ImpossibleStateError extends RuntimeException {
+    public ImpossibleStateError(String msg) {
+        super(msg);
+    }
+
+    public ImpossibleStateError(String msg, Throwable e) {
+        super(msg, e);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/InvalidOutputSchema.java b/src/main/java/org/apache/mahout/pig/InvalidOutputSchema.java
new file mode 100644
index 0000000..b642096
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/InvalidOutputSchema.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import java.io.IOException;
+
+public class InvalidOutputSchema extends IOException {
+    public InvalidOutputSchema(String msg) {
+        super(msg);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/LogisticRegression.java b/src/main/java/org/apache/mahout/pig/LogisticRegression.java
new file mode 100644
index 0000000..ff6953d
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/LogisticRegression.java
@@ -0,0 +1,341 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.io.Closeables;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.mahout.classifier.sgd.L1;
+import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.VectorWritable;
+import org.apache.pig.Accumulator;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.util.UDFContext;
+
+import java.io.*;
+import java.nio.file.Files;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * Trains a logistic regression model entirely in memory using the simplest learning algorithm from Mahout.
+ * <p/>
+ * A number of options can be supplied in a configuration string passed to the constructor.  These options
+ * are of the form name=value and options are separated by commas.  Whitespace around option names and around
+ * values is not significant.  Spaces inside values are preserved.
+ * <p/>
+ * The model to be trained can be read from a file in order to allow learning to continue at a previous
+ * stopping point or the particulars of a new model to be trained from scratch can be specified using the
+ * options.  Training data can be held in-memory or written to a temporary file.
+ * <p/>
+ * The following options can be used to load a pre-existing model:
+ * <p/>
+ * <ul>
+ * <li>model</li>
+ * <li>categories</li>
+ * </ul>
+ * <p/>
+ * The following options can be used with a newly created model:
+ * <p/>
+ * <ul>
+ * <li>categories - specifies a list of values that the target variable can take on.  This list should be space
+ * separated and given in the same order as when the model is later used.  (required)</li>
+ * <li>features - the number of elements in the feature vectors to be given to the learning algorithm.   (required)</li>
+ * <li>decayExponent - specifies how quickly the learning rate will decay.  Optional with default value of 0.5.
+ * Note that per-term annealing still has effect.</li>
+ * <li>lambda - specifies how much regularization constant is used.  Optional with default of 1e-5.</li>
+ * <li>stepOffset - slows down the decay of the learning rate at first if set greater than zero.  Default is 10</li>
+ * <li>learningRate - initial value of the learning rate.  Default is 1.</li>
+ * </ul>
+ * <p/>
+ * The following options can be used with any model:
+ * <p/>
+ * <ul>
+ * <li>inMemory - if "true" then training examples are kept in-memory and used in a random order.  If "false" then
+ * a temporary file is used to hold training examples and the order they are used is fixed by the order they are
+ * given to this UDF.  The default is "true".</li>
+ * <li>iterations - the number of iterations through the training data that are to be taken.</li>
+ * </ul>
+ */
+public class LogisticRegression extends EvalFunc<DataByteArray> implements Accumulator<DataByteArray> {
+    private List<String> categories;
+    OnlineLogisticRegression model;
+    List<Example> trainingData = Lists.newArrayList();
+    private int iterations;
+    private boolean inMemory;
+    private File tmpFile;
+
+    public LogisticRegression(String modelParams) throws IOException {
+        Splitter onComma = Splitter.on(",").trimResults().omitEmptyStrings();
+        Splitter onEquals = Splitter.on("=").trimResults();
+        Splitter onSpaces = Splitter.on(" ");
+        Joiner withSpaces = Joiner.on(" ");
+
+        Map<String, String> options = Maps.newHashMap();
+
+        for (String option : onComma.split(modelParams)) {
+            List<String> values = Lists.newArrayList(onEquals.split(option));
+            options.put(values.get(0), values.get(1));
+        }
+
+        if (options.containsKey("model")) {
+            if (options.containsKey("categories")) {
+                categories = Lists.newArrayList(onSpaces.split(options.get("categories")));
+                Configuration conf = UDFContext.getUDFContext().getJobConf();
+                model = PolymorphicWritable.read(FileSystem.get(conf).open(new Path(options.get("model"))), OnlineLogisticRegression.class);
+                options.remove("model");
+                options.remove(("categories"));
+            } else {
+                throw new BadClassifierSpecException("Must specify \"categories\" if pre-existing model is used");
+            }
+        } else {
+            if (options.containsKey("categories") && options.containsKey("features")) {
+                categories = Lists.newArrayList(onSpaces.split(options.get("categories")));
+                if (categories.size() < 2) {
+                    throw new BadClassifierSpecException("Must have more than one target category.  Remember that categories is a space separated list");
+                }
+                model = new OnlineLogisticRegression(categories.size(), Integer.parseInt(options.get("features")), new L1());
+                options.remove("categories");
+                options.remove("features");
+            } else {
+                throw new BadClassifierSpecException("Must specify previous model location using \"file\" or supply \"categories\" and \"features\"");
+            }
+
+            if (options.containsKey("decayExponent")) {
+                model.decayExponent(Double.parseDouble(options.get("decayExponent")));
+                options.remove("decayExponent");
+            }
+
+            if (options.containsKey("lambda")) {
+                model.lambda(Double.parseDouble(options.get("lambda")));
+                options.remove("lambda");
+            }
+
+            if (options.containsKey("stepOffset")) {
+                model.stepOffset(Integer.parseInt(options.get("stepOffset")));
+                options.remove("stepOffset");
+            }
+
+            if (options.containsKey("learningRate")) {
+                model.learningRate(Double.parseDouble(options.get("learningRate")));
+                options.remove("learningRate");
+            }
+        }
+
+        iterations = options.containsKey("iterations") ? Integer.parseInt(options.get("iterations")) : 1;
+        options.remove("iterations");
+
+        inMemory = options.containsKey("inMemory") ? Boolean.parseBoolean(options.get("inMemory")) : true;
+        options.remove("inMemory");
+
+        if (options.size() > 0) {
+            throw new BadClassifierSpecException("Extra options supplied: " + withSpaces.join(options.keySet()));
+        }
+
+        if (!inMemory) {
+            tmpFile = Files.createTempFile("trainingData", "tmp").toFile();
+            tmpFile.deleteOnExit();
+        }
+    }
+
+    @Override
+    public DataByteArray exec(Tuple input) throws IOException {
+        addBagOfData((DataBag) input.get(0));
+        return getValue();
+    }
+
+    /**
+     * Pass tuples to the learning algorithm.  Each tuple should have two fields.  The first
+     * fields should correspond to one of the categories for the model and the second should
+     * be the encoded features for the training example.
+     *
+     * @param example A tuple containing a single field, which is a bag.  The bag will contain the set
+     *                of training examples being passed to the learning algorithm in this iteration.  Not all
+     *                training examples will be passed at once.
+     */
+    public void accumulate(Tuple example) throws IOException {
+        if (example.size() != 1) {
+            throw new IllegalArgumentException("Input to training algorithm should be a single bag containing tuples each with target and vector");
+        }
+        addBagOfData((DataBag) example.get(0));
+    }
+
+    private void addBagOfData(DataBag data) throws IOException {
+        if (inMemory) {
+            for (Tuple input : data) {
+                trainingData.add(new Example(categories.indexOf(input.get(0)), PigVector.fromBytes((DataByteArray) input.get(1))));
+            }
+        } else {
+            DataOutputStream out = new DataOutputStream(new FileOutputStream(tmpFile));
+            try {
+                for (Tuple input : data) {
+                    out.writeInt(categories.indexOf(input.get(0)));
+                    PolymorphicWritable.write(out, new VectorWritable(PigVector.fromBytes((DataByteArray) input.get(1))));
+                }
+            } finally {
+                out.close();
+            }
+        }
+    }
+
+    /**
+     * Called when all tuples from current key have been passed to accumulate.  This is where the
+     * actual training occurs.  We can't do it earlier unless iterations = 1 which is an unusual
+     * case.
+     *
+     * @return the trained model.
+     */
+    public DataByteArray getValue() {
+        for (int i = 0; i < iterations; i++) {
+            for (Example example : readInput()) {
+                model.train(example.getTarget(), example.getFeatures());
+            }
+        }
+
+        try {
+            ByteArrayOutputStream baos = new ByteArrayOutputStream();
+            DataOutputStream out = new DataOutputStream(baos);
+            PolymorphicWritable.write(out, new Classifier(categories, model));
+            out.close();
+
+            return new DataByteArray(baos.toByteArray());
+        } catch (IOException e) {
+            // should never happen
+            throw new ImpossibleStateError("Can't put results into tuple", e);
+        }
+    }
+
+    /**
+     * Called after getValue() to prepare processing for next key.
+     */
+    public void cleanup() {
+        if (tmpFile != null) {
+            tmpFile.delete();
+        }
+    }
+
+    public int getIterations() {
+        return iterations;
+    }
+
+    public boolean isInMemory() {
+        return inMemory;
+    }
+
+    public OnlineLogisticRegression getModel() {
+        return model;
+    }
+
+    private Iterable<Example> readInput() {
+        if (inMemory) {
+            return new Iterable<Example>() {
+                public Iterator<Example> iterator() {
+                    return new AbstractIterator<Example>() {
+                        int remainingExamples = trainingData.size();
+                        Random gen = new Random();
+
+                        @Override
+                        protected Example computeNext() {
+                            if (remainingExamples > 0) {
+                                remainingExamples--;
+                                return trainingData.get(gen.nextInt(trainingData.size()));
+                            } else {
+                                return endOfData();
+                            }
+                        }
+                    };
+                }
+            };
+        } else {
+            return new Iterable<Example>() {
+                public Iterator<Example> iterator() {
+                    try {
+                        return new AbstractIterator<Example>() {
+                            DataInputStream in = new DataInputStream(new FileInputStream(tmpFile));
+
+                            @Override
+                            protected Example computeNext() {
+                                int target;
+                                try {
+                                    target = in.readInt();
+                                } catch (EOFException e) {
+                                    Closeables.closeQuietly(in);
+                                    return endOfData();
+                                } catch (IOException e) {
+                                    Closeables.closeQuietly(in);
+                                    throw new TrainingDataException("Error reading training data", e);
+                                }
+                                try {
+                                    return new Example(target, PolymorphicWritable.read(in, VectorWritable.class));
+                                } catch (EOFException e) {
+                                    Closeables.closeQuietly(in);
+                                    throw new TrainingDataException("Premature EOF while reading training data", e);
+                                } catch (IOException e) {
+                                    Closeables.closeQuietly(in);
+                                    throw new TrainingDataException("Error reading training data", e);
+                                }
+                            }
+                        };
+                    } catch (FileNotFoundException e) {
+                        throw new TrainingDataException("Could not training data file", e);
+                    }
+                }
+            };
+        }
+    }
+
+    private static class Example {
+        int target;
+        Vector features;
+
+        public Example(int target, Vector v) {
+            this.target = target;
+            this.features = v;
+        }
+
+        public Example(int target, VectorWritable v) {
+            this(target, v.get());
+        }
+
+        public int getTarget() {
+            return target;
+        }
+
+        public Vector getFeatures() {
+            return features;
+        }
+    }
+
+    private static class TrainingDataException extends RuntimeException {
+        public TrainingDataException(String msg, Throwable e) {
+            super(msg, e);
+        }
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/LogisticRegressionEval.java b/src/main/java/org/apache/mahout/pig/LogisticRegressionEval.java
new file mode 100644
index 0000000..31cd43f
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/LogisticRegressionEval.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.Text;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import org.apache.mahout.common.Pair;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirIterator;
+import org.apache.mahout.math.Vector;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.util.UDFContext;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Evaluates a logistic regression classifier
+ */
+public class LogisticRegressionEval extends EvalFunc<String> {
+    private Classifier model;
+    private String location;
+    private String key;
+
+    public LogisticRegressionEval(String modelParams) throws IOException {
+        System.out.printf("Model params = %s\n", modelParams);
+
+        Splitter onComma = Splitter.on(",").trimResults().omitEmptyStrings();
+        Splitter onEquals = Splitter.on("=").trimResults();
+        Joiner withSpaces = Joiner.on(" ");
+
+        Map<String, String> options = Maps.newHashMap();
+
+        for (String option : onComma.split(modelParams)) {
+            List<String> values = Lists.newArrayList(onEquals.split(option));
+            options.put(values.get(0), values.get(1));
+        }
+
+        if (options.containsKey("sequence")) {
+            location = options.get("sequence");
+            if (options.containsKey("key")) {
+                key = options.get("key");
+                options.remove("sequence");
+                options.remove(("key"));
+            } else {
+                throw new BadClassifierSpecException("Must specify key for model in a sequence file");
+            }
+        } else if (options.containsKey("file")) {
+            Configuration conf = UDFContext.getUDFContext().getJobConf();
+            final FSDataInputStream in = FileSystem.get(conf).open(new Path(options.get("file")));
+            try {
+                model = PolymorphicWritable.read(in, Classifier.class);
+            } finally {
+                in.close();
+            }
+            options.remove("file");
+
+        } else {
+            throw new BadClassifierSpecException("Must specify existing model");
+        }
+
+        if (options.size() > 0) {
+            throw new BadClassifierSpecException("Extra options supplied: " + withSpaces.join(options.keySet()));
+        }
+    }
+
+
+    /**
+     * This callback method must be implemented by all subclasses. This
+     * is the method that will be invoked on every Tuple of a given dataset.
+     * Since the dataset may be divided up in a variety of ways the programmer
+     * should not make assumptions about state that is maintained between
+     * invocations of this method.
+     *
+     * @param input the Tuple to be processed.
+     * @return result, of type T.
+     * @throws java.io.IOException
+     */
+    @Override
+    public String exec(Tuple input) throws IOException {
+        if (model == null) {
+            Path sequence = new Path(location);
+            Configuration conf = UDFContext.getUDFContext().getJobConf();
+            sequence = sequence.makeQualified(FileSystem.get(conf));
+            System.out.printf("Model file = %s\n", sequence);
+            SequenceFileDirIterator<Text, Classifier> x = new SequenceFileDirIterator<Text, Classifier>(sequence, PathType.GLOB, new PathFilter() {
+                public boolean accept(Path path) {
+                    System.out.printf("Scanning %s\n", path);
+                    return !path.toString().startsWith("_");
+                }
+            }, null, true, conf);
+            int n = 0;
+            while (x.hasNext()) {
+                Pair<Text, Classifier> pair = x.next();
+                n++;
+                if (pair.getFirst().toString().equals(key)) {
+                    System.out.printf("Found model with categories = %s\n", pair.getSecond().getCategories());
+                    model = pair.getSecond();
+                    break;
+                }
+            }
+            if (model == null) {
+                throw new IOException(String.format("Can't find model with correct key = %s out of %d searched\n", key, n));
+            }
+            x.close();
+        }
+        final Vector instance = PigVector.fromBytes(((DataByteArray) input.get(0)));
+        final int target = model.getModel().classifyFull(instance).maxValueIndex();
+        return model.getCategories().get(target);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/MessageLoader.java b/src/main/java/org/apache/mahout/pig/MessageLoader.java
new file mode 100644
index 0000000..713d492
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/MessageLoader.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Reads messages stored one per file such as in the 20 newsgroups data.
+ */
+public class MessageLoader extends LoadFunc {
+    private RecordReader<LongWritable, Text> reader;
+    private String location;
+
+    @Override
+    public void setLocation(String location, Job job) throws IOException {
+        this.location = location;
+        FileInputFormat.setInputPaths(job, this.location);
+    }
+
+    @Override
+    public InputFormat getInputFormat() throws IOException {
+        return new TextInputFormat() {
+
+            @Override
+            protected boolean isSplitable(JobContext context, Path file) {
+                return false;
+            }
+
+            @Override
+            public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) {
+                location = ((FileSplit) split).getPath().toString();
+                return super.createRecordReader(split, context);    //To change body of overridden methods use File | Settings | File Templates.
+            }
+        };
+    }
+
+    /**
+     * Stores our reader so we can get bytes.
+     */
+    @Override
+    public void prepareToRead(RecordReader reader, PigSplit split) throws IOException {
+        //noinspection unchecked
+        this.reader = reader;
+    }
+
+    /**
+     * Retrieves a message.  The resulting tuple has the directory name, the file name,
+     * the subject line and the text.
+     */
+    @Override
+    public Tuple getNext() throws IOException {
+        try {
+            String subject = "";
+            while (reader.nextKeyValue()) {
+                final Text line = reader.getCurrentValue();
+                if (line.find("Subject: ") == 0) {
+                    subject = line.toString().replace("Subject: ", "");
+                }
+                if (line.getLength() == 0) {
+                    break;
+                }
+            }
+
+            List<String> contents = Lists.newArrayList();
+            while (reader.nextKeyValue()) {
+                final String line = reader.getCurrentValue().toString();
+                if (line.equals("--")) {
+                    break;
+                }
+                if (!line.startsWith(">")) {
+                    contents.add(line);
+                }
+            }
+            if (subject.length() == 0 && contents.size() == 0) {
+                return null;
+            }
+            Tuple r = TupleFactory.getInstance().newTuple();
+
+            // directory without leading path
+            r.append(location.replaceAll("/[^/]*$", "").replaceAll(".*/", ""));
+            // file name without any directory
+            r.append(location.replaceAll(".*/", ""));
+            // subject line
+            r.append(subject);
+            // all the rest of the text
+            r.append(Joiner.on(" ").join(contents));
+
+            return r;
+        } catch (InterruptedException e) {
+            throw new ImpossibleStateError("Interrupted!");
+        }
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/PigModelStorage.java b/src/main/java/org/apache/mahout/pig/PigModelStorage.java
new file mode 100644
index 0000000..79bfdd6
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/PigModelStorage.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import org.apache.pig.ResourceSchema;
+import org.apache.pig.StoreFunc;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+
+/**
+ * Stores models in files named by a well known prefix and the reduce key.
+ */
+public class PigModelStorage extends StoreFunc {
+    private RecordWriter output;
+
+    public PigModelStorage() {
+    }
+
+
+    /**
+     * Set the schema for data to be stored.  This will be called on the
+     * front end during planning if the store is associated with a schema.
+     * A Store function should implement this function to
+     * check that a given schema is acceptable to it.  For example, it
+     * can check that the correct partition keys are included;
+     * a storage function to be written directly to an OutputFormat can
+     * make sure the schema will translate in a well defined way.  Default implementation
+     * is a no-op.
+     *
+     * @param s to be checked
+     * @throws java.io.IOException if this schema is not acceptable.  It should include
+     *                             a detailed error message indicating what is wrong with the schema.
+     */
+    @Override
+    public void checkSchema(ResourceSchema s) throws IOException {
+        if (s.getFields()[1].getType() != DataType.BYTEARRAY || s.getFields()[0].getType() == DataType.BYTEARRAY) {
+            throw new InvalidOutputSchema(String.format("Want a key with a string format and binary model for model output but got %s and %s",
+                    DataType.findTypeName(s.getFields()[0].getType()), DataType.findTypeName(s.getFields()[1].getType())));
+        }
+    }
+
+    /**
+     * Return the OutputFormat associated with StoreFunc.  This will be called
+     * on the front end during planning and on the backend during
+     * execution.
+     *
+     * @return the {@link org.apache.hadoop.mapreduce.OutputFormat} associated with StoreFunc
+     * @throws java.io.IOException if an exception occurs while constructing the
+     *                             OutputFormat
+     */
+    @Override
+    public OutputFormat getOutputFormat() throws IOException {
+        return new SequenceFileOutputFormat();
+    }
+
+    /**
+     * Communicate to the storer the location where the data needs to be stored.
+     * The location string passed to the {@link org.apache.pig.StoreFunc} here is the
+     * return value of {@link org.apache.pig.StoreFunc#relToAbsPathForStoreLocation(String, org.apache.hadoop.fs.Path)}
+     *
+     * @param location Location returned by
+     *                 {@link org.apache.pig.StoreFunc#relToAbsPathForStoreLocation(String, org.apache.hadoop.fs.Path)}
+     * @param job      The {@link org.apache.hadoop.mapreduce.Job} object
+     * @throws java.io.IOException if the location is not valid.
+     */
+    @Override
+    public void setStoreLocation(String location, Job job) throws IOException {
+        FileOutputFormat.setOutputPath(job, new Path(location));
+        job.setOutputKeyClass(Text.class);
+        job.setOutputValueClass(Classifier.class);
+    }
+
+    /**
+     * Initialize StoreFunc to write data.  This will be called during
+     * execution on the backend before the call to putNext.
+     *
+     * @param writer RecordWriter to use.
+     * @throws java.io.IOException if an exception occurs during initialization
+     */
+    @Override
+    public void prepareToWrite(RecordWriter writer) throws IOException {
+        this.output = writer;
+    }
+
+    /**
+     * Write a tuple to the data store.
+     *
+     * @param t the tuple to store.
+     * @throws java.io.IOException if an exception occurs during the write
+     */
+    @Override
+    public void putNext(Tuple t) throws IOException {
+        try {
+            Classifier r;
+            try {
+                DataInputStream in = new DataInputStream(new ByteArrayInputStream(((DataByteArray) t.get(1)).get()));
+                r = PolymorphicWritable.read(in, Classifier.class);
+                in.close();
+            } catch (IOException e) {
+                throw new ImpossibleStateError("Can't have error in BAIS", e);
+            }
+
+            //noinspection unchecked
+            output.write(new Text(DataType.toString(t.get(0))), r);
+        } catch (InterruptedException e) {
+            throw new ImpossibleStateError("Interrupted operation ... don't know what to do", e);
+        }
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/PigVector.java b/src/main/java/org/apache/mahout/pig/PigVector.java
new file mode 100644
index 0000000..d10fc6d
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/PigVector.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.VectorWritable;
+import org.apache.pig.data.DataByteArray;
+
+import java.io.*;
+
+/**
+ * Wraps a pig tuple around a Mahout vector.
+ */
+public class PigVector {
+    public static DataByteArray toBytes(Vector v1) {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos);
+        try {
+            new VectorWritable(v1).write(dos);
+            dos.close();
+        } catch (IOException e) {
+            throw new ImpossibleStateError("Can't have IO error on BAOS", e);
+        }
+        return new DataByteArray(baos.toByteArray());
+    }
+
+    public static Vector fromBytes(DataByteArray data) {
+        try {
+            VectorWritable r = new VectorWritable();
+            r.readFields(new DataInputStream(new ByteArrayInputStream(data.get())));
+            return r.get();
+        } catch (IOException e) {
+            throw new ImpossibleStateError("Can't have error in BAIS", e);
+        }
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/ArgumentEncoder.java b/src/main/java/org/apache/mahout/pig/encoders/ArgumentEncoder.java
new file mode 100644
index 0000000..770b304
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/ArgumentEncoder.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.util.Version;
+import org.apache.mahout.vectorizer.encoders.*;
+
+import java.lang.reflect.InvocationTargetException;
+
+public class ArgumentEncoder {
+    private int position;
+    private String name;
+    private FeatureVectorEncoder encoder;
+
+    public ArgumentEncoder(int position, String name) {
+        this.position = position;
+        this.name = name;
+    }
+
+    public static ArgumentEncoder newNumericEncoder(int position, String name) {
+        ArgumentEncoder r = new ArgumentEncoder(position, name);
+        r.encoder = new ContinuousValueEncoder(name);
+        return r;
+    }
+
+    public static ArgumentEncoder newWordEncoder(int position, String name) {
+        ArgumentEncoder r = new ArgumentEncoder(position, name);
+        r.encoder = new StaticWordValueEncoder(name);
+        return r;
+    }
+
+    public static ArgumentEncoder newTextEncoder(int position, String name) {
+        ArgumentEncoder r = new ArgumentEncoder(position, name);
+        r.encoder = new TextValueEncoder(name);
+        return r;
+    }
+
+    public static ArgumentEncoder newTextEncoder(int position, String name, String analyzerClass) throws SchemaParseException {
+        ArgumentEncoder r = new ArgumentEncoder(position, name);
+        LuceneTextValueEncoder enc = new LuceneTextValueEncoder(name);
+        Analyzer analyzer;
+        try {
+            analyzer = (Analyzer) Class.forName(analyzerClass).getConstructor(Version.class).newInstance(Version.LUCENE_31);
+        } catch (InstantiationException e) {
+            throw new SchemaParseException("Can't construct analyzer for class " + analyzerClass, e);
+        } catch (IllegalAccessException e) {
+            throw new SchemaParseException("Can't access analyzer class " + analyzerClass, e);
+        } catch (ClassNotFoundException e) {
+            throw new SchemaParseException("Can't find analyzer class " + analyzerClass, e);
+        } catch (NoSuchMethodException e) {
+            throw new SchemaParseException("Can't find constructor for analyzer class " + analyzerClass, e);
+        } catch (InvocationTargetException e) {
+            throw new SchemaParseException("Can't construct analyzer object " + analyzerClass, e);
+        }
+        enc.setAnalyzer(analyzer);
+        r.encoder = enc;
+        return r;
+    }
+
+    public FeatureVectorEncoder getEncoder() {
+        return encoder;
+    }
+
+    public int getPosition() {
+        return position;
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/DirectVector.java b/src/main/java/org/apache/mahout/pig/encoders/DirectVector.java
new file mode 100644
index 0000000..47a0f67
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/DirectVector.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+import org.apache.mahout.math.DenseVector;
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.pig.PigVector;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+
+import java.io.IOException;
+
+/**
+ * Encodes a set of numeric fields into distinct elements of a vector.
+ * <p/>
+ * No hashing is done and no dictionaries are used so we have exactly as many elements
+ * in the encoded vector as there are in the schema provided to the constructor of the
+ * encoder.
+ */
+public class DirectVector extends EvalFunc<DataByteArray> {
+    private final Splitter onCommas = Splitter.on(",").trimResults();
+    private final Iterable<String> fieldNames;
+    private final int dimension;
+
+    /**
+     * This class requires that a schema be provided in order to specify how
+     * to encode tuples.  The formula should be a list of field names separated by
+     * commas.
+     *
+     * @param schema A schema description for the arguments that will be given to the function later.
+     */
+    public DirectVector(String schema) {
+        fieldNames = onCommas.split(schema);
+        this.dimension = Iterables.size(fieldNames);
+    }
+
+    public DataByteArray exec(Tuple input) throws IOException {
+        if (input == null || input.size() != dimension) {
+            throw new IllegalArgumentException("Tuple doesn't have " + dimension + " elements");
+        } else {
+            Vector r = new DenseVector(dimension);
+            for (int i = 0; i < dimension; i++) {
+                final Object v = input.get(i);
+                if (v != null) {
+                    r.set(i, DataType.toDouble(v));
+                } else {
+                    r.set(i, Double.NaN);
+                }
+            }
+            return PigVector.toBytes(r);
+        }
+    }
+
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/EncodeVector.java b/src/main/java/org/apache/mahout/pig/encoders/EncodeVector.java
new file mode 100644
index 0000000..e1ddbb1
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/EncodeVector.java
@@ -0,0 +1,97 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import org.apache.mahout.math.RandomAccessSparseVector;
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.pig.ImpossibleStateError;
+import org.apache.mahout.pig.PigVector;
+import org.apache.mahout.vectorizer.encoders.ConstantValueEncoder;
+import org.apache.mahout.vectorizer.encoders.FeatureVectorEncoder;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.Tuple;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Encodes a tuple as a vector using the hashed vector representation.
+ */
+public class EncodeVector extends EvalFunc<DataByteArray> {
+    private int dimension;
+    private EncodingSpec spec;
+    private Map<String, ArgumentEncoder> encoderMap;
+    private final FeatureVectorEncoder constantEncoder = new ConstantValueEncoder("- bias -");
+    private int minimumTupleSize;
+
+    /**
+     * This class requires that a formula and a schema be provided in order to specify how
+     * to encode tuples.  The formula should be like an R formula such as is used by lm or glm.
+     * <p/>
+     * The formula can look like this
+     * <p/>
+     * x + y*z + 1
+     * <p/>
+     * This indicates that x, the interaction between x and y and a constant offset should be encoded.
+     * <p/>
+     * The schema looks generally like a pig schema except that only the following primitive types are recognized:
+     * <p/>
+     * numeric
+     * word
+     * text
+     * text(Analyzer)
+     *
+     * @param dimension Number of elements in the resulting encoded vectors
+     * @param formula   The R-like formula that describes the data to encode
+     * @param schema    A schema description for the arguments that will be given to the function later.
+     */
+    public EncodeVector(String dimension, String formula, String schema) {
+        this.dimension = Integer.parseInt(dimension);
+        encoderMap = Schema.parse(schema);
+        minimumTupleSize = 0;
+        for (ArgumentEncoder encoder : encoderMap.values()) {
+            minimumTupleSize = Math.max(minimumTupleSize, encoder.getPosition() + 1);
+        }
+        spec = Formula.parse(formula);
+    }
+
+    public DataByteArray exec(Tuple input) throws IOException {
+        if (input == null || input.size() == 0 || input.size() < minimumTupleSize) {
+            throw new IllegalArgumentException("Tuple doesn't have at least " + minimumTupleSize + " elements");
+        } else {
+            Vector r = new RandomAccessSparseVector(dimension);
+            for (Set<String> variables : spec.getVariables()) {
+                if (variables.size() > 1) {
+                    throw new UnsupportedOperationException("Can't encode interactions yet");
+                }
+                if (variables.size() == 0) {
+                    throw new ImpossibleStateError("No variables!");
+                }
+                final String var = variables.iterator().next();
+                if ("1".equals(var)) {
+                    constantEncoder.addToVector((byte[]) null, r);
+                } else {
+                    ArgumentEncoder encoder = encoderMap.get(var);
+                    encoder.getEncoder().addToVector(input.get(encoder.getPosition()).toString(), r);
+                }
+            }
+            return PigVector.toBytes(r);
+        }
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/EncodingSpec.java b/src/main/java/org/apache/mahout/pig/encoders/EncodingSpec.java
new file mode 100644
index 0000000..dd980b1
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/EncodingSpec.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
+
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * Describes the variables and interactions expressed by an encoding specification.
+ */
+public class EncodingSpec {
+    private Set<Set<String>> variables = Sets.newTreeSet(new Comparator<Set<String>>() {
+        public int compare(Set<String> s1, Set<String> s2) {
+            Iterator<String> i1 = s1.iterator();
+            Iterator<String> i2 = s2.iterator();
+            while (i1.hasNext() && i2.hasNext()) {
+                int r = i1.next().compareTo(i2.next());
+                if (r != 0) {
+                    return r;
+                }
+            }
+            if (i2.hasNext()) {
+                return -1;
+            } else if (i1.hasNext()) {
+                return 1;
+            } else {
+                return 0;
+            }
+        }
+    });
+
+    public EncodingSpec(String variable) {
+        variables.add(ImmutableSet.of(variable));
+    }
+
+    public EncodingSpec(EncodingSpec v1) {
+        this.variables.addAll(v1.getVariables());
+    }
+
+    public EncodingSpec(EncodingSpec v1, EncodingSpec v2) {
+        this.variables.addAll(v1.getVariables());
+        this.variables.addAll(v2.getVariables());
+        for (Set<String> x1 : v1.getVariables()) {
+            for (Set<String> x2 : v2.getVariables()) {
+                Set<String> both = Sets.newTreeSet(x1);
+                both.addAll(x2);
+                variables.add(both);
+            }
+        }
+    }
+
+    private void add(EncodingSpec spec) {
+        variables.addAll(spec.getVariables());
+    }
+
+    public Set<Set<String>> getVariables() {
+        return variables;
+    }
+
+    public static EncodingSpec add(EncodingSpec a, EncodingSpec b) {
+        EncodingSpec r = new EncodingSpec(a);
+        r.add(b);
+        return r;
+    }
+
+    public static EncodingSpec interact(EncodingSpec a, EncodingSpec b) {
+        return new EncodingSpec(a, b);
+    }
+
+    public static EncodingSpec pow(EncodingSpec a, int n) {
+        EncodingSpec r = new EncodingSpec(a);
+        if (n > r.getVariables().size()) {
+            n = r.getVariables().size();
+        }
+        for (int i = 1; i < n; i++) {
+            r = new EncodingSpec(r, a);
+        }
+        return r;
+    }
+
+    public static EncodingSpec atom(String s) {
+        return new EncodingSpec(s);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/Formula.java b/src/main/java/org/apache/mahout/pig/encoders/Formula.java
new file mode 100644
index 0000000..93dc6c7
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/Formula.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import org.antlr.runtime.ANTLRReaderStream;
+import org.antlr.runtime.CommonTokenStream;
+import org.antlr.runtime.RecognitionException;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.io.StringReader;
+
+public class Formula {
+    public static EncodingSpec parse(Reader reader) {
+        FormulaParser tokenParser;
+        try {
+            CommonTokenStream input = new CommonTokenStream(new FormulaLexer(new ANTLRReaderStream(reader)));
+            tokenParser = new FormulaParser(input);
+            FormulaParser.expression_return parserResult = tokenParser.expression();
+            return parserResult.r;
+        } catch (IOException e) {
+            throw new SchemaParseException("Cannot parse schema", e);
+        } catch (RecognitionException e) {
+            throw new SchemaParseException("Cannot parse schema", e);
+        }
+    }
+
+    public static EncodingSpec parse(String formula) {
+        final StringReader input = new StringReader(formula);
+        EncodingSpec r = parse(input);
+        input.close();
+        return r;
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/FormulaParseException.java b/src/main/java/org/apache/mahout/pig/encoders/FormulaParseException.java
new file mode 100644
index 0000000..60813a2
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/FormulaParseException.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+public class FormulaParseException extends RuntimeException {
+    public FormulaParseException(Throwable e, String msg) {
+        super(msg, e);
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/Schema.java b/src/main/java/org/apache/mahout/pig/encoders/Schema.java
new file mode 100644
index 0000000..c7c57e6
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/Schema.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import org.antlr.runtime.ANTLRReaderStream;
+import org.antlr.runtime.CommonTokenStream;
+import org.antlr.runtime.RecognitionException;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.io.StringReader;
+import java.util.Map;
+
+/**
+ * A schema specifies the names and types for positional parameters in a tuple.  These names
+ * and types are used when encoding tuples as vectors using PigVector.
+ * <p/>
+ * This class contains convenience routines that simplify the use of the schema parser classes.
+ * <p/>
+ * A schema consists of a comma separated list of variable specifications.  A variable
+ * specification contains a name plus optional absolute position and a type separated by a colon.
+ * The name follows the rules of java identifiers.  The optional absolute position is indicated
+ * by a dollar sign and a number.  The number indicates the 0-based position of the variable
+ * in any tuple being encoded.  Variables without positions are assigned consecutive positions
+ * starting at 0 in the tuple.
+ * <p/>
+ * The type specification can be numeric, word or text.  For text, you can supply an optional
+ * parenthesis surrounded fully qualified name of a Java class that implements the Analyzer
+ * interface from Lucene.  If you don't supply an analyzer, the text is tokenized on whilte-
+ * space boundaries.
+ * <p/>
+ * Here is an example of a schema:
+ * <pre>
+ * a:numeric, b$0:word, c$1:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text
+ * </pre>
+ * Here, argument 0 in the encoded tuple is encoded as a number and as a key word while
+ * argument 1 in the encoded tuple is encoded as text using two different methods for
+ * analysis.
+ */
+public class Schema {
+    public static Map<String, ArgumentEncoder> parse(Reader reader) {
+        SchemaParser tokenParser;
+        try {
+            CommonTokenStream input = new CommonTokenStream(new SchemaLexer(new ANTLRReaderStream(reader)));
+            tokenParser = new SchemaParser(input);
+            // parsing causes side effect on tokenParser
+            tokenParser.schema();
+        } catch (IOException e) {
+            throw new SchemaParseException("Cannot parse schema", e);
+        } catch (RecognitionException e) {
+            throw new SchemaParseException("Cannot parse schema", e);
+        }
+        return tokenParser.getEncoders();
+    }
+
+    public static Map<String, ArgumentEncoder> parse(String s) {
+        final StringReader input = new StringReader(s);
+        Map<String, ArgumentEncoder> r = parse(input);
+        input.close();
+        return r;
+    }
+}
diff --git a/src/main/java/org/apache/mahout/pig/encoders/SchemaParseException.java b/src/main/java/org/apache/mahout/pig/encoders/SchemaParseException.java
new file mode 100644
index 0000000..0f737f8
--- /dev/null
+++ b/src/main/java/org/apache/mahout/pig/encoders/SchemaParseException.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+public class SchemaParseException extends RuntimeException {
+    public SchemaParseException(String s, Throwable throwable) {
+        super(s, throwable);
+    }
+}
diff --git a/src/test/java/org/apache/mahout/pig/EncodeVectorTest.java b/src/test/java/org/apache/mahout/pig/EncodeVectorTest.java
new file mode 100644
index 0000000..a798ff4
--- /dev/null
+++ b/src/test/java/org/apache/mahout/pig/EncodeVectorTest.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.function.Functions;
+import org.apache.mahout.pig.encoders.EncodeVector;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.*;
+
+public class EncodeVectorTest {
+
+    @Test
+    public void testSimple() throws IOException {
+        EncodeVector ev = new EncodeVector("100", "1+x+y", "x:numeric, not:text, y:text");
+        Tuple t = TupleFactory.getInstance().newTuple(4);
+        t.set(0, 3.1);
+        t.set(1, "foo bar");
+        t.set(2, "");
+        t.set(3, null);
+
+        Vector v1 = PigVector.fromBytes(ev.exec(t));
+        assertEquals(3.1, v1.get(10), 0);
+        assertEquals(1.0, v1.get(39), 0);
+
+        t.set(2, "fob");
+
+        Vector pv = PigVector.fromBytes(ev.exec(t));
+        Vector v2 = pv.assign(v1, Functions.MINUS);
+        assertEquals(1, v2.get(40), 0);
+        assertEquals(1, v2.get(55), 0);
+
+        t.set(2, "lots of text for testing");
+
+        pv = PigVector.fromBytes(ev.exec(t));
+        Vector v3 = pv.assign(v1, Functions.MINUS);
+        assertEquals(10, v3.zSum(), 0);
+        assertEquals(1, v3.maxValue(), 1);
+
+        EncodeVector ev2 = new EncodeVector("100", "x+y", "x:numeric, not:text, y:text");
+        t.set(2, "");
+        pv = PigVector.fromBytes(ev2.exec(t));
+        Vector v4 = pv;
+        assertEquals(3.1, v1.get(10), 0);
+        assertEquals(3.1, v4.zSum(), 0);
+        assertEquals(3.1, v4.maxValue(), 1);
+    }
+
+    @Test
+    public void testBadIndex() throws IOException {
+        EncodeVector ev2 = new EncodeVector("100", "x+y", "x:numeric, not:text, y:text");
+        Tuple t = TupleFactory.getInstance().newTuple(2);
+        t.set(0, 3.1);
+        t.set(1, "foo bar");
+        try {
+            ev2.exec(t);
+            fail("Should have failed because tuple is too small");
+        } catch (IllegalArgumentException e) {
+            assertTrue(e.getMessage().contains("Tuple doesn't have at least"));
+        }
+    }
+}
diff --git a/src/test/java/org/apache/mahout/pig/EncodingSpecTest.java b/src/test/java/org/apache/mahout/pig/EncodingSpecTest.java
new file mode 100644
index 0000000..8787bbb
--- /dev/null
+++ b/src/test/java/org/apache/mahout/pig/EncodingSpecTest.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import org.antlr.runtime.ANTLRReaderStream;
+import org.antlr.runtime.CommonTokenStream;
+import org.antlr.runtime.RecognitionException;
+import org.apache.mahout.pig.encoders.EncodingSpec;
+import org.apache.mahout.pig.encoders.FormulaLexer;
+import org.apache.mahout.pig.encoders.FormulaParseException;
+import org.apache.mahout.pig.encoders.FormulaParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.io.StringReader;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class EncodingSpecTest {
+    @Test
+    public void testSum() throws RecognitionException, IOException {
+        EncodingSpec z = parse("a+c+b");
+        Assert.assertEquals("[[a], [b], [c]]", z.getVariables().toString());
+
+        try {
+            parse("a+$1+$2");
+            fail("Should have failed");
+        } catch (FormulaParseException e) {
+            assertTrue(e.getMessage().startsWith("Syntax error"));
+        }
+
+        z = parse("a + 1");
+        Assert.assertEquals("[[1], [a]]", z.getVariables().toString());
+    }
+
+    @Test
+    public void testWrongBias() throws RecognitionException, IOException {
+        try {
+            parse("a + 2");
+            fail("Should have failed due to invalid bias");
+        } catch (FormulaParseException e) {
+            assertTrue(e.getMessage().startsWith("Invalid bias"));
+        }
+    }
+
+    @Test
+    public void testProduct() throws RecognitionException, IOException {
+        EncodingSpec z = parse("a+c*b");
+        Assert.assertEquals("[[a], [b], [b, c], [c]]", z.getVariables().toString());
+        z = parse("(a+c)*(c+b+a)");
+        Assert.assertEquals("[[a], [a, b], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
+    }
+
+    @Test
+    public void testPower() throws RecognitionException, IOException {
+        EncodingSpec z = parse("(a+b+c)^2");
+        Assert.assertEquals("[[a], [a, b], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
+
+        z = parse("(a+b+c)^3");
+        Assert.assertEquals("[[a], [a, b], [a, b, c], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
+
+        z = parse("(a+b+c)^30");
+        Assert.assertEquals("[[a], [a, b], [a, b, c], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
+    }
+
+    private EncodingSpec parse(String s) throws IOException, RecognitionException {
+        Reader reader = new StringReader(s);
+        CommonTokenStream input = new CommonTokenStream(new FormulaLexer(new ANTLRReaderStream(reader)));
+        FormulaParser tokenParser = new FormulaParser(input);
+        FormulaParser.expression_return parserResult = tokenParser.expression();
+        reader.close();
+        return parserResult.r;
+    }
+
+}
diff --git a/src/test/java/org/apache/mahout/pig/LogisticRegressionTest.java b/src/test/java/org/apache/mahout/pig/LogisticRegressionTest.java
new file mode 100644
index 0000000..84f1e1e
--- /dev/null
+++ b/src/test/java/org/apache/mahout/pig/LogisticRegressionTest.java
@@ -0,0 +1,130 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import org.apache.mahout.math.DenseVector;
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.function.DoubleFunction;
+import org.apache.pig.data.*;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.List;
+import java.util.Random;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.fail;
+import static org.junit.Assert.assertTrue;
+
+public class LogisticRegressionTest {
+    @Test
+    public void testConstructor() throws IOException {
+        LogisticRegression lr = new LogisticRegression("categories = a b c, features=10000");
+        assertEquals(3, lr.getModel().numCategories());
+        assertEquals(10000, lr.getModel().numFeatures());
+        assertEquals(1, lr.getIterations());
+        assertTrue(lr.isInMemory());
+
+
+        lr = new LogisticRegression("categories = a b c, features=10000, decayExponent=0.3 ,stepOffset=123, learningRate=2.1, lambda=0.001, iterations=12, inMemory=false");
+        assertEquals(2.1 * Math.pow(123, -0.3), lr.getModel().currentLearningRate());
+        assertEquals(0.001, lr.getModel().getLambda());
+
+        try {
+            new LogisticRegression("categories = a , features=10000");
+            fail("Should have failed");
+        } catch (BadClassifierSpecException e) {
+            assertTrue("Single target category", e.getMessage().startsWith("Must have more than one target category"));
+        }
+
+        try {
+            new LogisticRegression("categories = a b, features=10000, x=3");
+            fail("Should have failed");
+        } catch (BadClassifierSpecException e) {
+            assertEquals("Extra options", "Extra options supplied: x", e.getMessage());
+        }
+
+        try {
+            new LogisticRegression("categories = a ");
+            fail("Should have failed");
+        } catch (BadClassifierSpecException e) {
+            assertEquals("No features", "Must specify previous model location using \"file\" or supply \"categories\" and \"features\"", e.getMessage());
+        }
+    }
+
+    @Test
+    public void testTraining() throws IOException {
+        DoubleFunction randomValue = new DoubleFunction() {
+            private Random gen = new Random(1);
+
+            public double apply(double arg1) {
+                return gen.nextGaussian();
+            }
+        };
+
+        // start with a random direction
+        Vector n = new DenseVector(4);
+        n.assign(randomValue);
+
+        // make up data.  result is 1 if in the same rough direction as n, else 0
+        String[] target = new String[]{"0", "1"};
+        DataBag examples = new DefaultDataBag();
+        for (int i = 0; i < 400; i++) {
+            Vector v = new DenseVector(4);
+            v.assign(randomValue);
+
+            Tuple x = new DefaultTuple();
+            x.append(target[v.dot(n) > 0 ? 1 : 0]);
+            x.append(PigVector.toBytes(v));
+            examples.add(x);
+        }
+        Tuple data = new DefaultTuple();
+        data.append(examples);
+
+        // train model.  training from tmp file allows absolute repeatability
+        LogisticRegression lr = new LogisticRegression("categories = 0 1, features=4, inMemory=false, iterations=5");
+        lr.accumulate(data);
+        DataByteArray r = lr.getValue();
+        DataInputStream in = new DataInputStream(new ByteArrayInputStream(r.get()));
+        Classifier c = PolymorphicWritable.read(in, Classifier.class);
+        assertEquals(2, c.getCategories().size());
+        assertEquals("0", c.getCategories().get(0));
+        assertEquals("1", c.getCategories().get(1));
+        OnlineLogisticRegression model = c.getModel();
+        assertEquals(lr.getModel().currentLearningRate(), model.currentLearningRate(), 1e-10);
+        in.close();
+
+        // with that many data points, model should point in the same direction as the original vector
+        Vector v = model.getBeta().viewRow(0);
+        double z = n.dot(v) / (n.norm(2) * v.norm(2));
+        assertEquals(1.0, z, 1e-2);
+
+        // just for grins, we should check whether the model actually computes the correct values
+        List<String> categories = ImmutableList.of("0", "1");
+        for (Tuple example : examples) {
+            double score = model.classifyScalar(PigVector.fromBytes((DataByteArray) example.get(1)));
+            int actual = categories.indexOf(example.get(0));
+            score = score * actual + (1 - score) * (1 - actual);
+            assertTrue(score > 0.4);
+        }
+    }
+}
diff --git a/src/test/java/org/apache/mahout/pig/SchemaTest.java b/src/test/java/org/apache/mahout/pig/SchemaTest.java
new file mode 100644
index 0000000..ef879ee
--- /dev/null
+++ b/src/test/java/org/apache/mahout/pig/SchemaTest.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig;
+
+import org.antlr.runtime.RecognitionException;
+import org.apache.mahout.pig.encoders.ArgumentEncoder;
+import org.apache.mahout.pig.encoders.Schema;
+import org.apache.mahout.pig.encoders.SchemaParseException;
+import org.apache.mahout.vectorizer.encoders.ContinuousValueEncoder;
+import org.apache.mahout.vectorizer.encoders.LuceneTextValueEncoder;
+import org.apache.mahout.vectorizer.encoders.StaticWordValueEncoder;
+import org.apache.mahout.vectorizer.encoders.TextValueEncoder;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+public class SchemaTest {
+    @Test
+    public void testBasics() throws RecognitionException, IOException {
+        Map<String, ArgumentEncoder> r = Schema.parse("a:numeric, b:word, c:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text");
+        assertEquals(4, r.size());
+        assertTrue(r.get("a").getEncoder() instanceof ContinuousValueEncoder);
+        assertTrue(r.get("b").getEncoder() instanceof StaticWordValueEncoder);
+        assertTrue(r.get("c").getEncoder() instanceof LuceneTextValueEncoder);
+        assertTrue(r.get("d").getEncoder() instanceof TextValueEncoder);
+    }
+
+    @Test
+    public void testFancyPositioning() throws RecognitionException, IOException {
+        Map<String, ArgumentEncoder> r = Schema.parse("a:numeric, b$0:word, c$1:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text");
+        assertEquals(4, r.size());
+        assertTrue(r.get("a").getEncoder() instanceof ContinuousValueEncoder);
+        assertTrue(r.get("b").getEncoder() instanceof StaticWordValueEncoder);
+        assertTrue(r.get("c").getEncoder() instanceof LuceneTextValueEncoder);
+        assertTrue(r.get("d").getEncoder() instanceof TextValueEncoder);
+
+        assertEquals(0, r.get("a").getPosition());
+        assertEquals(0, r.get("b").getPosition());
+        assertEquals(1, r.get("c").getPosition());
+        assertEquals(1, r.get("d").getPosition());
+    }
+
+    @Test
+    public void testMultipleDefinition() throws RecognitionException, IOException {
+        try {
+            Schema.parse("a:numeric, b:word, a:text, d:text");
+            fail("Should have thrown syntax error");
+        } catch (SchemaParseException e) {
+            assertTrue(e.getMessage().contains("multiply defined"));
+        }
+    }
+
+    @Test
+    public void testSyntaxError() throws RecognitionException, IOException {
+        try {
+            Schema.parse("a, b:word, a:text, d:text");
+            fail("Should have thrown syntax error");
+        } catch (SchemaParseException e) {
+            assertTrue(e.getMessage().contains("Syntax error"));
+        }
+    }
+}
diff --git a/src/test/java/org/apache/mahout/pig/encoders/DirectVectorTest.java b/src/test/java/org/apache/mahout/pig/encoders/DirectVectorTest.java
new file mode 100644
index 0000000..3c9aa3d
--- /dev/null
+++ b/src/test/java/org/apache/mahout/pig/encoders/DirectVectorTest.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2014 Ted Dunning
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.pig.encoders;
+
+import org.apache.mahout.math.Vector;
+import org.apache.mahout.pig.PigVector;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static junit.framework.Assert.*;
+
+public class DirectVectorTest {
+    @Test
+    public void testEncoding() throws IOException {
+        DirectVector dv = new DirectVector("a,b,c");
+        final Tuple input = TupleFactory.getInstance().newTuple();
+        input.append(1.0);
+        input.append(2);
+        input.append(3L);
+        Vector v = PigVector.fromBytes(dv.exec(input));
+        assertEquals(3, v.size());
+        assertEquals(1.0, v.get(0), 0);
+        assertEquals(2.0, v.get(1), 0);
+        assertEquals(3.0, v.get(2), 0);
+    }
+
+    @Test
+    public void testWrongSize() throws IOException {
+        DirectVector dv = new DirectVector("a,b,c");
+        final Tuple input = TupleFactory.getInstance().newTuple();
+        input.append(1.0);
+        input.append(2);
+        try {
+            dv.exec(input);
+            fail("Should have noticed short tuple");
+        } catch (IllegalArgumentException e) {
+            assertEquals("Tuple doesn't have 3 elements", e.getMessage());
+        }
+    }
+
+
+    @Test
+    public void testNullAsNaN() throws IOException {
+        DirectVector dv = new DirectVector("a,b,c");
+        final Tuple input = TupleFactory.getInstance().newTuple();
+        input.append(1.0);
+        input.append(2);
+        input.append(null);
+
+        Vector v = PigVector.fromBytes(dv.exec(input));
+        assertEquals(3, v.size());
+        assertEquals(1.0, v.get(0), 0);
+        assertEquals(2.0, v.get(1), 0);
+        assertTrue(Double.isNaN(v.get(2)));
+    }
+}
diff --git a/src/test/pig/encoderTest.pig b/src/test/pig/encoderTest.pig
new file mode 100644
index 0000000..2b01a94
--- /dev/null
+++ b/src/test/pig/encoderTest.pig
@@ -0,0 +1,8 @@
+register 'target/pig-vector-1.0-jar-with-dependencies.jar';
+define EncodeVector org.apache.mahout.pig.encoders.EncodeVector('10','x+y+1', 'x:numeric, y:numeric');
+define DirectVector org.apache.mahout.pig.encoders.DirectVector('x1,x2,x3');
+
+a = load '/Users/tdunning/Downloads/NNBench.csv' using PigStorage(',') as (x1:int, x2:int, x3:int);
+b = foreach a generate EncodeVector(*), DirectVector(*);
+dump b;         
+store b into 'foo.bin' using BinStorage;

Fixed code: .idea/*
target/*
*.iml
This code provides the ability to encode data in Pig using the hashed encoding capabilities of Mahout.  It also allows
training of logistic regression models.

To try this out, start by compiling and installing mahout in your local repository:

    cd ~/Apache
    git clone https://github.com/apache/mahout.git
    cd mahout
    mvn install -DskipTests

Then do the same with pig-vector

    cd ~/Apache
    git clone git@github.com:tdunning/pig-vector.git
    cd pig-vector
    mvn package

Now you have a comprehensive jar that includes pig-vector and Mahout.  This jar is big (24M) but it has a lot of stuff on its mind.

Now you can write a program in pig that uses Mahout vectors.

To do this, start off by registering the pig-vector jar:

    register 'target/pig-vector-1.0-jar-with-dependencies.jar';

Then define an encoding function.  The arguments to the constructor are 

- the dimension of the resulting vectors (100,000 to 1,000,000 is a more common range than the 10 showed here)

- a description of the variables to be included in the encoding.    The interactions currently don't work, but all the rest does.

- the schema of the tuples that pig will pass together with their data types.

There is a lot of flexibility here in terms of tokenizing text, reusing fields with different encodings and interaction variables.

    define EncodeVector org.apache.mahout.pig.encoders.EncodeVector('10','x+y+1', 'x:numeric, y:numeric, z:numeric');

The other important types of data are word (for categorical data) and text.  If you just say "text" is the type, then
whitespace is used for tokenization.  You can also add a Lucene 3.1 tokenizer in parentheses if you want something fancier
in the way of tokenization.  The encoder treats word fields as strings with no processing whatsoever.

The formula is not a normal arithmetic formula, but is more akin to an R formula as used in regression.  In this style,
you list the variables you want to include, joined by +'s.  You can add 1 if you want a constant terms which is very important
for most generalized linear classifiers.  You can specify interactions using * or you can add all interactions of a group
of terms by using exponentiation.  That allows something like (x+y+z)^3 which includes all first, second and third order
variables.  The syntax doesn't yet allow the ':' operator of R and when it comes time to encoding no interactions are implemented
yet, regardless of syntax support.

Then we just need data to encode.  Here I read an output file that has some silly data in it.

    a = load '/Users/tdunning/Downloads/NNBench.csv' using PigStorage(',') as (x1:int, x2:int, x3:int);

and here is code that encodes all of the data (that we care about):

    b = foreach a generate 1 as key, EncodeVector(*) as v;

Note that the schema for the tuple has fields x1, x2, x3 while the schema for the encoding has x, y, z.
Since the schema of the tuple is not visible to the UDF, we can't always avoid this kind of mismatch except by careful
coding of the pig code.  Note also how the encoding of the vector (defined by the formula "x+y+1") includes fields x1
and x2 and a bias term but excludes x3.

You can look at the output

    dump b;         

The output is in binary form so you can't see it, but it has two non-zero fields that encode x1, two that
encode x2 and a single field that encodes the bias. With only 10 dimensional output vectors, some of these may
overlap and cause a larger value.

Or you can write the vector out in a sequence file by using the elephant-bird package from twitter which is
included in pig-vector:

    store b into 'vectors.dat' using com.twitter.elephantbird.pig.store.SequenceFileStorage (
       '-c com.twitter.elephantbird.pig.util.IntWritableConverter',
       '-c com.twitter.elephantbird.pig.util.GenericWritableConverter -t org.apache.mahout.math.VectorWritable'
    );

This data isn't very interesting because the key for every vector is the constant value 1, but you can see the
pattern.  Note how the IntWritableConverter is used to specify the type of the key and the GenericWritableConverter
is used to specify how to write the vector value.  The -t option is required so that we can specify what the
actual type of is for the data we want to write.

Vectors encoded using DirectVector or EncodeVector are passed around in pig as bytearray's.  If they are passed
to a learning algorithm, they have to be converted into real vector objects in order to be passed to the learning
algorithm.  When writing them out to a file, we can just use the bytes as is with the help of the
GenericWritableConverter class.

To train a model, we use the org.apache.mahout.pig.LogisticRegression UDF.  The constructor for this function
requires a set of options to specify how it should run.  Here is a sample:

    define train org.apache.mahout.pig.LogisticRegression('iterations=5, inMemory=true, features=100000, categories=alt.atheism comp.sys.mac.hardware rec.motorcycles sci.electronics talk.politics.guns comp.graphics comp.windows.x rec.sport.baseball sci.med talk.politics.mideast comp.os.ms-windows.misc misc.forsale rec.sport.hockey sci.space talk.politics.misc comp.sys.ibm.pc.hardware rec.autos sci.crypt soc.religion.christian talk.religion.misc');

This very long line specifies the number of learning iterations (5), that the training vectors should be kept
in memory and that the feature vectors are have 100,000 elements.  The categories option is a space delimited
list of the categories and must match the strings that are passed into the learning algorithm as the target variable.

Here is a complete training script:

    register 'target/pig-vector-1.0-jar-with-dependencies.jar';
    define encodeVector org.apache.mahout.pig.encoders.EncodeVector('100000', 'subject+body', 'group:word, article:numeric, subject:text, body:text');
    define train org.apache.mahout.pig.LogisticRegression('iterations=5, inMemory=true, features=100000, categories=alt.atheism comp.sys.mac.hardware rec.motorcycles sci.electronics talk.politics.guns comp.graphics comp.windows.x rec.sport.baseball sci.med talk.politics.mideast comp.os.ms-windows.misc misc.forsale rec.sport.hockey sci.space talk.politics.misc comp.sys.ibm.pc.hardware rec.autos sci.crypt soc.religion.christian talk.religion.misc');

    rm model.dat

    /* read the data */
    docs = load '20news-bydate-train/*/*' using org.apache.mahout.pig.MessageLoader()
         as (newsgroup, id:int, subject, body);
    /* encode as vectors, retain the target variable and the feature vector */
    vectors = foreach docs generate newsgroup, encodeVector(*) as v;
    describe vectors;

    /* put the training data in a single bag.  We could train multiple models this way */
    grouped = group vectors all;

    /* train the actual model.  The key is bogus to satisfy the sequence vector format. */
    model = foreach grouped generate 1 as key, train(vectors) as model;

    /* the trained model is passed to use as a bytearray so we just pass it on out.  The classifier
       class just contains the list of target valeus and the OnlineLogisticRegression object itself. */
    store model into 'model.dat' using com.twitter.elephantbird.pig.store.SequenceFileStorage (
       '-c com.twitter.elephantbird.pig.util.IntWritableConverter',
       '-c com.twitter.elephantbird.pig.util.GenericWritableConverter -t org.apache.mahout.pig.Classifier'
    );

Send questions and suggestions to ted.dunning@gmail.com

<?xml version="1.0" encoding="UTF-8"?>
<!--
  ~ Copyright 2014 Ted Dunning
  ~
  ~ Licensed under the Apache License, Version 2.0 (the "License");
  ~ you may not use this file except in compliance with the License.
  ~ You may obtain a copy of the License at
  ~
  ~     http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  -->

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>pig-vector</groupId>
    <artifactId>pig-vector</artifactId>
    <version>1.0</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.pig</groupId>
            <artifactId>pig</artifactId>
            <version>0.9.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.twitter.elephantbird</groupId>
            <artifactId>elephant-bird-core</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.mahout</groupId>
            <artifactId>mahout-core</artifactId>
            <version>0.8</version>
        </dependency>
        <dependency>
            <groupId>org.antlr</groupId>
            <artifactId>antlr</artifactId>
            <version>3.4</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.antlr</groupId>
                <artifactId>antlr3-maven-plugin</artifactId>
                <version>3.4</version>

                <executions>
                    <execution>
                        <id>antlr-generate</id>
                        <!-- this is used for inheritance merges -->
                        <phase>generate-sources</phase>
                        <!-- bind to the packaging phase -->
                        <goals>
                            <goal>antlr</goal>
                        </goals>
                    </execution>
                </executions>

            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>1.6</source>
                    <target>1.6</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>
grammar Formula;

options {
  output = AST;
}

@header {
package org.apache.mahout.pig.encoders;
}

@lexer::header {
package org.apache.mahout.pig.encoders;
}

@members {
  public void reportError(RecognitionException e) {
    if (e instanceof FailedPredicateException) {
      throw new FormulaParseException(e, "Invalid bias");
    } else {
      throw new FormulaParseException(e, "Syntax error in schema: ");
    }
  }
}

expression returns [EncodingSpec r]: s=sumExpression EOF  {$r = $s.r;};

sumExpression returns [EncodingSpec r]:
    f1=factor {$r = $f1.r;}
       (PLUS f2=factor {$r = EncodingSpec.add($r, $f2.r);})*
    ;

factor returns [EncodingSpec r]:
    a1=argument {$r = $a1.r;} (TIMES a2=argument {$r = EncodingSpec.interact($r, $a2.r);})*  ;
	
argument returns [EncodingSpec r]:
    LPAREN s1=sumExpression  RPAREN  {$r = $s1.r;}
    (UP_ARROW n = INTEGER {$r = EncodingSpec.pow($s1.r, Integer.parseInt($n.getText()));})?
    | ID {$r = EncodingSpec.atom($ID.text);}
    | bias {$r = EncodingSpec.atom("1");};

bias : i = INTEGER {"1".equals(i.getText())}?;

PLUS:       '+';
TIMES:      '*';
LPAREN:     '(';
RPAREN:     ')';
UP_ARROW:   '^';
INTEGER:	('0'..'9')+;
ID:         ('a'..'z'|'A'..'Z'|'_') ('a'..'z'|'A'..'Z'|'0'..'9'|'_')*;
DOLLAR:     '$';
WS       :           (' '|'\t'|'\f'|'\n'|'\r')+{ $channel=HIDDEN; };

grammar Schema;

options {
  output = AST;
}

@header {
package org.apache.mahout.pig.encoders;

import com.google.common.collect.Maps;
import java.util.Map;
import java.util.HashMap;
}

@lexer::header {
package org.apache.mahout.pig.encoders;
}

@members {       
       private int position = 0;
       private Map<String, ArgumentEncoder> encoders = Maps.newHashMap();
     
       public Map<String, ArgumentEncoder> getEncoders() {
          return encoders;
       }

       public void reportError(RecognitionException e) {
         if (e instanceof FailedPredicateException) {
           throw new SchemaParseException("Variable multiply defined", e);
         } else {
           throw new SchemaParseException("Syntax error in schema: ", e);
         }
       }

}

schema 	:	{position = 0;} variable (COMMA! variable)*;

variable	:
          ID^ COLON! t=type[$ID.text, position++]
            {encoders.get($ID.text) == null}?
            {encoders.put($ID.text, $t.encoder);}
          | ID^ DOLLAR INTEGER COLON! t=type[$ID.text, Integer.parseInt($INTEGER.text)]
            {encoders.get($ID.text) == null}?
            {encoders.put($ID.text, $t.encoder);};

type [String name, int position] returns [ArgumentEncoder encoder]:	
          NUMERIC^ {$encoder = ArgumentEncoder.newNumericEncoder(position, $name);}
          | WORD^ {$encoder = ArgumentEncoder.newWordEncoder(position, $name);}
          | TEXT^ {$encoder = ArgumentEncoder.newTextEncoder(position, $name);}
          | TEXT^ LPAREN! className RPAREN! {$encoder = ArgumentEncoder.newTextEncoder(position, $name, $className.text);}
          ;
          
className	:	ID (DOT ID)*;

NUMERIC	:	'numeric';
WORD	:	'word';
TEXT	:	'text';
INTEGER:	('0'..'9')+;

LPAREN	:	'(';
RPAREN	:	')';
COLON	:	':';
ID  :	('a'..'z'|'A'..'Z'|'_') ('a'..'z'|'A'..'Z'|'0'..'9'|'_')*
    ;
DOT	:	'.';
COMMA	:	',';
DOLLAR	:	'$';

WS  :   ( ' '
        | '\t'
        | '\r'
        | '\n'
        ) {$channel=HIDDEN;}
    ;

/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

/**
 * Signal an error in classifier construction.
 */
public class BadClassifierSpecException extends RuntimeException {
    public BadClassifierSpecException(String msg, Throwable e) {
        super(msg, e);
    }

    public BadClassifierSpecException(String msg) {
        super(msg);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.pig.impl.util.UDFContext;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;

/**
 * Writable wrapper for a classifier and a list of target categories.
 */
public class Classifier implements Writable {
    private List<String> categories;
    private OnlineLogisticRegression model;

    public Classifier() {
        categories = Lists.newArrayList();
    }

    public Classifier(List<String> categories, OnlineLogisticRegression model) {
        this.categories = Lists.newArrayList(categories);
        this.model = model;
    }

    public Classifier(String file) {
        try {
            Configuration conf = UDFContext.getUDFContext().getJobConf();
            final FSDataInputStream in = FileSystem.get(conf).open(new Path(file));
            try {
                this.readFields(in);
            } finally {
                in.close();
            }
        } catch (FileNotFoundException e) {
            throw new BadClassifierSpecException("Can't find file to read model from: " + file, e);
        } catch (IOException e) {
            throw new BadClassifierSpecException("Error while reading model from: " + file, e);
        }
    }

    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeInt(categories.size());
        for (String category : categories) {
            new Text(category).write(dataOutput);
        }
        PolymorphicWritable.write(dataOutput, model);
    }

    public void readFields(DataInput dataInput) throws IOException {
        int n = dataInput.readInt();
        for (int i = 0; i < n; i++) {
            Text tmp = new Text();
            tmp.readFields(dataInput);
            categories.add(tmp.toString());
        }
        model = PolymorphicWritable.read(dataInput, OnlineLogisticRegression.class);
    }

    public List<String> getCategories() {
        return categories;
    }

    public OnlineLogisticRegression getModel() {
        return model;
    }

}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

/**
 * Indicates an "impossible" condition has been detected that prevents further progress.
 */
public class ImpossibleStateError extends RuntimeException {
    public ImpossibleStateError(String msg) {
        super(msg);
    }

    public ImpossibleStateError(String msg, Throwable e) {
        super(msg, e);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import java.io.IOException;

public class InvalidOutputSchema extends IOException {
    public InvalidOutputSchema(String msg) {
        super(msg);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.io.Closeables;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.mahout.classifier.sgd.L1;
import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.pig.Accumulator;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.UDFContext;

import java.io.*;
import java.nio.file.Files;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * Trains a logistic regression model entirely in memory using the simplest learning algorithm from Mahout.
 * <p/>
 * A number of options can be supplied in a configuration string passed to the constructor.  These options
 * are of the form name=value and options are separated by commas.  Whitespace around option names and around
 * values is not significant.  Spaces inside values are preserved.
 * <p/>
 * The model to be trained can be read from a file in order to allow learning to continue at a previous
 * stopping point or the particulars of a new model to be trained from scratch can be specified using the
 * options.  Training data can be held in-memory or written to a temporary file.
 * <p/>
 * The following options can be used to load a pre-existing model:
 * <p/>
 * <ul>
 * <li>model</li>
 * <li>categories</li>
 * </ul>
 * <p/>
 * The following options can be used with a newly created model:
 * <p/>
 * <ul>
 * <li>categories - specifies a list of values that the target variable can take on.  This list should be space
 * separated and given in the same order as when the model is later used.  (required)</li>
 * <li>features - the number of elements in the feature vectors to be given to the learning algorithm.   (required)</li>
 * <li>decayExponent - specifies how quickly the learning rate will decay.  Optional with default value of 0.5.
 * Note that per-term annealing still has effect.</li>
 * <li>lambda - specifies how much regularization constant is used.  Optional with default of 1e-5.</li>
 * <li>stepOffset - slows down the decay of the learning rate at first if set greater than zero.  Default is 10</li>
 * <li>learningRate - initial value of the learning rate.  Default is 1.</li>
 * </ul>
 * <p/>
 * The following options can be used with any model:
 * <p/>
 * <ul>
 * <li>inMemory - if "true" then training examples are kept in-memory and used in a random order.  If "false" then
 * a temporary file is used to hold training examples and the order they are used is fixed by the order they are
 * given to this UDF.  The default is "true".</li>
 * <li>iterations - the number of iterations through the training data that are to be taken.</li>
 * </ul>
 */
public class LogisticRegression extends EvalFunc<DataByteArray> implements Accumulator<DataByteArray> {
    private List<String> categories;
    OnlineLogisticRegression model;
    List<Example> trainingData = Lists.newArrayList();
    private int iterations;
    private boolean inMemory;
    private File tmpFile;

    public LogisticRegression(String modelParams) throws IOException {
        Splitter onComma = Splitter.on(",").trimResults().omitEmptyStrings();
        Splitter onEquals = Splitter.on("=").trimResults();
        Splitter onSpaces = Splitter.on(" ");
        Joiner withSpaces = Joiner.on(" ");

        Map<String, String> options = Maps.newHashMap();

        for (String option : onComma.split(modelParams)) {
            List<String> values = Lists.newArrayList(onEquals.split(option));
            options.put(values.get(0), values.get(1));
        }

        if (options.containsKey("model")) {
            if (options.containsKey("categories")) {
                categories = Lists.newArrayList(onSpaces.split(options.get("categories")));
                Configuration conf = UDFContext.getUDFContext().getJobConf();
                model = PolymorphicWritable.read(FileSystem.get(conf).open(new Path(options.get("model"))), OnlineLogisticRegression.class);
                options.remove("model");
                options.remove(("categories"));
            } else {
                throw new BadClassifierSpecException("Must specify \"categories\" if pre-existing model is used");
            }
        } else {
            if (options.containsKey("categories") && options.containsKey("features")) {
                categories = Lists.newArrayList(onSpaces.split(options.get("categories")));
                if (categories.size() < 2) {
                    throw new BadClassifierSpecException("Must have more than one target category.  Remember that categories is a space separated list");
                }
                model = new OnlineLogisticRegression(categories.size(), Integer.parseInt(options.get("features")), new L1());
                options.remove("categories");
                options.remove("features");
            } else {
                throw new BadClassifierSpecException("Must specify previous model location using \"file\" or supply \"categories\" and \"features\"");
            }

            if (options.containsKey("decayExponent")) {
                model.decayExponent(Double.parseDouble(options.get("decayExponent")));
                options.remove("decayExponent");
            }

            if (options.containsKey("lambda")) {
                model.lambda(Double.parseDouble(options.get("lambda")));
                options.remove("lambda");
            }

            if (options.containsKey("stepOffset")) {
                model.stepOffset(Integer.parseInt(options.get("stepOffset")));
                options.remove("stepOffset");
            }

            if (options.containsKey("learningRate")) {
                model.learningRate(Double.parseDouble(options.get("learningRate")));
                options.remove("learningRate");
            }
        }

        iterations = options.containsKey("iterations") ? Integer.parseInt(options.get("iterations")) : 1;
        options.remove("iterations");

        inMemory = options.containsKey("inMemory") ? Boolean.parseBoolean(options.get("inMemory")) : true;
        options.remove("inMemory");

        if (options.size() > 0) {
            throw new BadClassifierSpecException("Extra options supplied: " + withSpaces.join(options.keySet()));
        }

        if (!inMemory) {
            tmpFile = Files.createTempFile("trainingData", "tmp").toFile();
            tmpFile.deleteOnExit();
        }
    }

    @Override
    public DataByteArray exec(Tuple input) throws IOException {
        addBagOfData((DataBag) input.get(0));
        return getValue();
    }

    /**
     * Pass tuples to the learning algorithm.  Each tuple should have two fields.  The first
     * fields should correspond to one of the categories for the model and the second should
     * be the encoded features for the training example.
     *
     * @param example A tuple containing a single field, which is a bag.  The bag will contain the set
     *                of training examples being passed to the learning algorithm in this iteration.  Not all
     *                training examples will be passed at once.
     */
    public void accumulate(Tuple example) throws IOException {
        if (example.size() != 1) {
            throw new IllegalArgumentException("Input to training algorithm should be a single bag containing tuples each with target and vector");
        }
        addBagOfData((DataBag) example.get(0));
    }

    private void addBagOfData(DataBag data) throws IOException {
        if (inMemory) {
            for (Tuple input : data) {
                trainingData.add(new Example(categories.indexOf(input.get(0)), PigVector.fromBytes((DataByteArray) input.get(1))));
            }
        } else {
            DataOutputStream out = new DataOutputStream(new FileOutputStream(tmpFile));
            try {
                for (Tuple input : data) {
                    out.writeInt(categories.indexOf(input.get(0)));
                    PolymorphicWritable.write(out, new VectorWritable(PigVector.fromBytes((DataByteArray) input.get(1))));
                }
            } finally {
                out.close();
            }
        }
    }

    /**
     * Called when all tuples from current key have been passed to accumulate.  This is where the
     * actual training occurs.  We can't do it earlier unless iterations = 1 which is an unusual
     * case.
     *
     * @return the trained model.
     */
    public DataByteArray getValue() {
        for (int i = 0; i < iterations; i++) {
            for (Example example : readInput()) {
                model.train(example.getTarget(), example.getFeatures());
            }
        }

        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            DataOutputStream out = new DataOutputStream(baos);
            PolymorphicWritable.write(out, new Classifier(categories, model));
            out.close();

            return new DataByteArray(baos.toByteArray());
        } catch (IOException e) {
            // should never happen
            throw new ImpossibleStateError("Can't put results into tuple", e);
        }
    }

    /**
     * Called after getValue() to prepare processing for next key.
     */
    public void cleanup() {
        if (tmpFile != null) {
            tmpFile.delete();
        }
    }

    public int getIterations() {
        return iterations;
    }

    public boolean isInMemory() {
        return inMemory;
    }

    public OnlineLogisticRegression getModel() {
        return model;
    }

    private Iterable<Example> readInput() {
        if (inMemory) {
            return new Iterable<Example>() {
                public Iterator<Example> iterator() {
                    return new AbstractIterator<Example>() {
                        int remainingExamples = trainingData.size();
                        Random gen = new Random();

                        @Override
                        protected Example computeNext() {
                            if (remainingExamples > 0) {
                                remainingExamples--;
                                return trainingData.get(gen.nextInt(trainingData.size()));
                            } else {
                                return endOfData();
                            }
                        }
                    };
                }
            };
        } else {
            return new Iterable<Example>() {
                public Iterator<Example> iterator() {
                    try {
                        return new AbstractIterator<Example>() {
                            DataInputStream in = new DataInputStream(new FileInputStream(tmpFile));

                            @Override
                            protected Example computeNext() {
                                int target;
                                try {
                                    target = in.readInt();
                                } catch (EOFException e) {
                                    Closeables.closeQuietly(in);
                                    return endOfData();
                                } catch (IOException e) {
                                    Closeables.closeQuietly(in);
                                    throw new TrainingDataException("Error reading training data", e);
                                }
                                try {
                                    return new Example(target, PolymorphicWritable.read(in, VectorWritable.class));
                                } catch (EOFException e) {
                                    Closeables.closeQuietly(in);
                                    throw new TrainingDataException("Premature EOF while reading training data", e);
                                } catch (IOException e) {
                                    Closeables.closeQuietly(in);
                                    throw new TrainingDataException("Error reading training data", e);
                                }
                            }
                        };
                    } catch (FileNotFoundException e) {
                        throw new TrainingDataException("Could not training data file", e);
                    }
                }
            };
        }
    }

    private static class Example {
        int target;
        Vector features;

        public Example(int target, Vector v) {
            this.target = target;
            this.features = v;
        }

        public Example(int target, VectorWritable v) {
            this(target, v.get());
        }

        public int getTarget() {
            return target;
        }

        public Vector getFeatures() {
            return features;
        }
    }

    private static class TrainingDataException extends RuntimeException {
        public TrainingDataException(String msg, Throwable e) {
            super(msg, e);
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.io.Text;
import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.mahout.common.Pair;
import org.apache.mahout.common.iterator.sequencefile.PathType;
import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirIterator;
import org.apache.mahout.math.Vector;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.UDFContext;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * Evaluates a logistic regression classifier
 */
public class LogisticRegressionEval extends EvalFunc<String> {
    private Classifier model;
    private String location;
    private String key;

    public LogisticRegressionEval(String modelParams) throws IOException {
        System.out.printf("Model params = %s\n", modelParams);

        Splitter onComma = Splitter.on(",").trimResults().omitEmptyStrings();
        Splitter onEquals = Splitter.on("=").trimResults();
        Joiner withSpaces = Joiner.on(" ");

        Map<String, String> options = Maps.newHashMap();

        for (String option : onComma.split(modelParams)) {
            List<String> values = Lists.newArrayList(onEquals.split(option));
            options.put(values.get(0), values.get(1));
        }

        if (options.containsKey("sequence")) {
            location = options.get("sequence");
            if (options.containsKey("key")) {
                key = options.get("key");
                options.remove("sequence");
                options.remove(("key"));
            } else {
                throw new BadClassifierSpecException("Must specify key for model in a sequence file");
            }
        } else if (options.containsKey("file")) {
            Configuration conf = UDFContext.getUDFContext().getJobConf();
            final FSDataInputStream in = FileSystem.get(conf).open(new Path(options.get("file")));
            try {
                model = PolymorphicWritable.read(in, Classifier.class);
            } finally {
                in.close();
            }
            options.remove("file");

        } else {
            throw new BadClassifierSpecException("Must specify existing model");
        }

        if (options.size() > 0) {
            throw new BadClassifierSpecException("Extra options supplied: " + withSpaces.join(options.keySet()));
        }
    }


    /**
     * This callback method must be implemented by all subclasses. This
     * is the method that will be invoked on every Tuple of a given dataset.
     * Since the dataset may be divided up in a variety of ways the programmer
     * should not make assumptions about state that is maintained between
     * invocations of this method.
     *
     * @param input the Tuple to be processed.
     * @return result, of type T.
     * @throws java.io.IOException
     */
    @Override
    public String exec(Tuple input) throws IOException {
        if (model == null) {
            Path sequence = new Path(location);
            Configuration conf = UDFContext.getUDFContext().getJobConf();
            sequence = sequence.makeQualified(FileSystem.get(conf));
            System.out.printf("Model file = %s\n", sequence);
            SequenceFileDirIterator<Text, Classifier> x = new SequenceFileDirIterator<Text, Classifier>(sequence, PathType.GLOB, new PathFilter() {
                public boolean accept(Path path) {
                    System.out.printf("Scanning %s\n", path);
                    return !path.toString().startsWith("_");
                }
            }, null, true, conf);
            int n = 0;
            while (x.hasNext()) {
                Pair<Text, Classifier> pair = x.next();
                n++;
                if (pair.getFirst().toString().equals(key)) {
                    System.out.printf("Found model with categories = %s\n", pair.getSecond().getCategories());
                    model = pair.getSecond();
                    break;
                }
            }
            if (model == null) {
                throw new IOException(String.format("Can't find model with correct key = %s out of %d searched\n", key, n));
            }
            x.close();
        }
        final Vector instance = PigVector.fromBytes(((DataByteArray) input.get(0)));
        final int target = model.getModel().classifyFull(instance).maxValueIndex();
        return model.getCategories().get(target);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.pig.LoadFunc;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import java.io.IOException;
import java.util.List;

/**
 * Reads messages stored one per file such as in the 20 newsgroups data.
 */
public class MessageLoader extends LoadFunc {
    private RecordReader<LongWritable, Text> reader;
    private String location;

    @Override
    public void setLocation(String location, Job job) throws IOException {
        this.location = location;
        FileInputFormat.setInputPaths(job, this.location);
    }

    @Override
    public InputFormat getInputFormat() throws IOException {
        return new TextInputFormat() {

            @Override
            protected boolean isSplitable(JobContext context, Path file) {
                return false;
            }

            @Override
            public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) {
                location = ((FileSplit) split).getPath().toString();
                return super.createRecordReader(split, context);    //To change body of overridden methods use File | Settings | File Templates.
            }
        };
    }

    /**
     * Stores our reader so we can get bytes.
     */
    @Override
    public void prepareToRead(RecordReader reader, PigSplit split) throws IOException {
        //noinspection unchecked
        this.reader = reader;
    }

    /**
     * Retrieves a message.  The resulting tuple has the directory name, the file name,
     * the subject line and the text.
     */
    @Override
    public Tuple getNext() throws IOException {
        try {
            String subject = "";
            while (reader.nextKeyValue()) {
                final Text line = reader.getCurrentValue();
                if (line.find("Subject: ") == 0) {
                    subject = line.toString().replace("Subject: ", "");
                }
                if (line.getLength() == 0) {
                    break;
                }
            }

            List<String> contents = Lists.newArrayList();
            while (reader.nextKeyValue()) {
                final String line = reader.getCurrentValue().toString();
                if (line.equals("--")) {
                    break;
                }
                if (!line.startsWith(">")) {
                    contents.add(line);
                }
            }
            if (subject.length() == 0 && contents.size() == 0) {
                return null;
            }
            Tuple r = TupleFactory.getInstance().newTuple();

            // directory without leading path
            r.append(location.replaceAll("/[^/]*$", "").replaceAll(".*/", ""));
            // file name without any directory
            r.append(location.replaceAll(".*/", ""));
            // subject line
            r.append(subject);
            // all the rest of the text
            r.append(Joiner.on(" ").join(contents));

            return r;
        } catch (InterruptedException e) {
            throw new ImpossibleStateError("Interrupted!");
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;

/**
 * Stores models in files named by a well known prefix and the reduce key.
 */
public class PigModelStorage extends StoreFunc {
    private RecordWriter output;

    public PigModelStorage() {
    }


    /**
     * Set the schema for data to be stored.  This will be called on the
     * front end during planning if the store is associated with a schema.
     * A Store function should implement this function to
     * check that a given schema is acceptable to it.  For example, it
     * can check that the correct partition keys are included;
     * a storage function to be written directly to an OutputFormat can
     * make sure the schema will translate in a well defined way.  Default implementation
     * is a no-op.
     *
     * @param s to be checked
     * @throws java.io.IOException if this schema is not acceptable.  It should include
     *                             a detailed error message indicating what is wrong with the schema.
     */
    @Override
    public void checkSchema(ResourceSchema s) throws IOException {
        if (s.getFields()[1].getType() != DataType.BYTEARRAY || s.getFields()[0].getType() == DataType.BYTEARRAY) {
            throw new InvalidOutputSchema(String.format("Want a key with a string format and binary model for model output but got %s and %s",
                    DataType.findTypeName(s.getFields()[0].getType()), DataType.findTypeName(s.getFields()[1].getType())));
        }
    }

    /**
     * Return the OutputFormat associated with StoreFunc.  This will be called
     * on the front end during planning and on the backend during
     * execution.
     *
     * @return the {@link org.apache.hadoop.mapreduce.OutputFormat} associated with StoreFunc
     * @throws java.io.IOException if an exception occurs while constructing the
     *                             OutputFormat
     */
    @Override
    public OutputFormat getOutputFormat() throws IOException {
        return new SequenceFileOutputFormat();
    }

    /**
     * Communicate to the storer the location where the data needs to be stored.
     * The location string passed to the {@link org.apache.pig.StoreFunc} here is the
     * return value of {@link org.apache.pig.StoreFunc#relToAbsPathForStoreLocation(String, org.apache.hadoop.fs.Path)}
     *
     * @param location Location returned by
     *                 {@link org.apache.pig.StoreFunc#relToAbsPathForStoreLocation(String, org.apache.hadoop.fs.Path)}
     * @param job      The {@link org.apache.hadoop.mapreduce.Job} object
     * @throws java.io.IOException if the location is not valid.
     */
    @Override
    public void setStoreLocation(String location, Job job) throws IOException {
        FileOutputFormat.setOutputPath(job, new Path(location));
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Classifier.class);
    }

    /**
     * Initialize StoreFunc to write data.  This will be called during
     * execution on the backend before the call to putNext.
     *
     * @param writer RecordWriter to use.
     * @throws java.io.IOException if an exception occurs during initialization
     */
    @Override
    public void prepareToWrite(RecordWriter writer) throws IOException {
        this.output = writer;
    }

    /**
     * Write a tuple to the data store.
     *
     * @param t the tuple to store.
     * @throws java.io.IOException if an exception occurs during the write
     */
    @Override
    public void putNext(Tuple t) throws IOException {
        try {
            Classifier r;
            try {
                DataInputStream in = new DataInputStream(new ByteArrayInputStream(((DataByteArray) t.get(1)).get()));
                r = PolymorphicWritable.read(in, Classifier.class);
                in.close();
            } catch (IOException e) {
                throw new ImpossibleStateError("Can't have error in BAIS", e);
            }

            //noinspection unchecked
            output.write(new Text(DataType.toString(t.get(0))), r);
        } catch (InterruptedException e) {
            throw new ImpossibleStateError("Interrupted operation ... don't know what to do", e);
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.pig.data.DataByteArray;

import java.io.*;

/**
 * Wraps a pig tuple around a Mahout vector.
 */
public class PigVector {
    public static DataByteArray toBytes(Vector v1) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        try {
            new VectorWritable(v1).write(dos);
            dos.close();
        } catch (IOException e) {
            throw new ImpossibleStateError("Can't have IO error on BAOS", e);
        }
        return new DataByteArray(baos.toByteArray());
    }

    public static Vector fromBytes(DataByteArray data) {
        try {
            VectorWritable r = new VectorWritable();
            r.readFields(new DataInputStream(new ByteArrayInputStream(data.get())));
            return r.get();
        } catch (IOException e) {
            throw new ImpossibleStateError("Can't have error in BAIS", e);
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.util.Version;
import org.apache.mahout.vectorizer.encoders.*;

import java.lang.reflect.InvocationTargetException;

public class ArgumentEncoder {
    private int position;
    private String name;
    private FeatureVectorEncoder encoder;

    public ArgumentEncoder(int position, String name) {
        this.position = position;
        this.name = name;
    }

    public static ArgumentEncoder newNumericEncoder(int position, String name) {
        ArgumentEncoder r = new ArgumentEncoder(position, name);
        r.encoder = new ContinuousValueEncoder(name);
        return r;
    }

    public static ArgumentEncoder newWordEncoder(int position, String name) {
        ArgumentEncoder r = new ArgumentEncoder(position, name);
        r.encoder = new StaticWordValueEncoder(name);
        return r;
    }

    public static ArgumentEncoder newTextEncoder(int position, String name) {
        ArgumentEncoder r = new ArgumentEncoder(position, name);
        r.encoder = new TextValueEncoder(name);
        return r;
    }

    public static ArgumentEncoder newTextEncoder(int position, String name, String analyzerClass) throws SchemaParseException {
        ArgumentEncoder r = new ArgumentEncoder(position, name);
        LuceneTextValueEncoder enc = new LuceneTextValueEncoder(name);
        Analyzer analyzer;
        try {
            analyzer = (Analyzer) Class.forName(analyzerClass).getConstructor(Version.class).newInstance(Version.LUCENE_31);
        } catch (InstantiationException e) {
            throw new SchemaParseException("Can't construct analyzer for class " + analyzerClass, e);
        } catch (IllegalAccessException e) {
            throw new SchemaParseException("Can't access analyzer class " + analyzerClass, e);
        } catch (ClassNotFoundException e) {
            throw new SchemaParseException("Can't find analyzer class " + analyzerClass, e);
        } catch (NoSuchMethodException e) {
            throw new SchemaParseException("Can't find constructor for analyzer class " + analyzerClass, e);
        } catch (InvocationTargetException e) {
            throw new SchemaParseException("Can't construct analyzer object " + analyzerClass, e);
        }
        enc.setAnalyzer(analyzer);
        r.encoder = enc;
        return r;
    }

    public FeatureVectorEncoder getEncoder() {
        return encoder;
    }

    public int getPosition() {
        return position;
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import com.google.common.base.Splitter;
import com.google.common.collect.Iterables;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.pig.PigVector;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;

import java.io.IOException;

/**
 * Encodes a set of numeric fields into distinct elements of a vector.
 * <p/>
 * No hashing is done and no dictionaries are used so we have exactly as many elements
 * in the encoded vector as there are in the schema provided to the constructor of the
 * encoder.
 */
public class DirectVector extends EvalFunc<DataByteArray> {
    private final Splitter onCommas = Splitter.on(",").trimResults();
    private final Iterable<String> fieldNames;
    private final int dimension;

    /**
     * This class requires that a schema be provided in order to specify how
     * to encode tuples.  The formula should be a list of field names separated by
     * commas.
     *
     * @param schema A schema description for the arguments that will be given to the function later.
     */
    public DirectVector(String schema) {
        fieldNames = onCommas.split(schema);
        this.dimension = Iterables.size(fieldNames);
    }

    public DataByteArray exec(Tuple input) throws IOException {
        if (input == null || input.size() != dimension) {
            throw new IllegalArgumentException("Tuple doesn't have " + dimension + " elements");
        } else {
            Vector r = new DenseVector(dimension);
            for (int i = 0; i < dimension; i++) {
                final Object v = input.get(i);
                if (v != null) {
                    r.set(i, DataType.toDouble(v));
                } else {
                    r.set(i, Double.NaN);
                }
            }
            return PigVector.toBytes(r);
        }
    }

}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.pig.ImpossibleStateError;
import org.apache.mahout.pig.PigVector;
import org.apache.mahout.vectorizer.encoders.ConstantValueEncoder;
import org.apache.mahout.vectorizer.encoders.FeatureVectorEncoder;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;

import java.io.IOException;
import java.util.Map;
import java.util.Set;

/**
 * Encodes a tuple as a vector using the hashed vector representation.
 */
public class EncodeVector extends EvalFunc<DataByteArray> {
    private int dimension;
    private EncodingSpec spec;
    private Map<String, ArgumentEncoder> encoderMap;
    private final FeatureVectorEncoder constantEncoder = new ConstantValueEncoder("- bias -");
    private int minimumTupleSize;

    /**
     * This class requires that a formula and a schema be provided in order to specify how
     * to encode tuples.  The formula should be like an R formula such as is used by lm or glm.
     * <p/>
     * The formula can look like this
     * <p/>
     * x + y*z + 1
     * <p/>
     * This indicates that x, the interaction between x and y and a constant offset should be encoded.
     * <p/>
     * The schema looks generally like a pig schema except that only the following primitive types are recognized:
     * <p/>
     * numeric
     * word
     * text
     * text(Analyzer)
     *
     * @param dimension Number of elements in the resulting encoded vectors
     * @param formula   The R-like formula that describes the data to encode
     * @param schema    A schema description for the arguments that will be given to the function later.
     */
    public EncodeVector(String dimension, String formula, String schema) {
        this.dimension = Integer.parseInt(dimension);
        encoderMap = Schema.parse(schema);
        minimumTupleSize = 0;
        for (ArgumentEncoder encoder : encoderMap.values()) {
            minimumTupleSize = Math.max(minimumTupleSize, encoder.getPosition() + 1);
        }
        spec = Formula.parse(formula);
    }

    public DataByteArray exec(Tuple input) throws IOException {
        if (input == null || input.size() == 0 || input.size() < minimumTupleSize) {
            throw new IllegalArgumentException("Tuple doesn't have at least " + minimumTupleSize + " elements");
        } else {
            Vector r = new RandomAccessSparseVector(dimension);
            for (Set<String> variables : spec.getVariables()) {
                if (variables.size() > 1) {
                    throw new UnsupportedOperationException("Can't encode interactions yet");
                }
                if (variables.size() == 0) {
                    throw new ImpossibleStateError("No variables!");
                }
                final String var = variables.iterator().next();
                if ("1".equals(var)) {
                    constantEncoder.addToVector((byte[]) null, r);
                } else {
                    ArgumentEncoder encoder = encoderMap.get(var);
                    encoder.getEncoder().addToVector(input.get(encoder.getPosition()).toString(), r);
                }
            }
            return PigVector.toBytes(r);
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;

import java.util.Comparator;
import java.util.Iterator;
import java.util.Set;

/**
 * Describes the variables and interactions expressed by an encoding specification.
 */
public class EncodingSpec {
    private Set<Set<String>> variables = Sets.newTreeSet(new Comparator<Set<String>>() {
        public int compare(Set<String> s1, Set<String> s2) {
            Iterator<String> i1 = s1.iterator();
            Iterator<String> i2 = s2.iterator();
            while (i1.hasNext() && i2.hasNext()) {
                int r = i1.next().compareTo(i2.next());
                if (r != 0) {
                    return r;
                }
            }
            if (i2.hasNext()) {
                return -1;
            } else if (i1.hasNext()) {
                return 1;
            } else {
                return 0;
            }
        }
    });

    public EncodingSpec(String variable) {
        variables.add(ImmutableSet.of(variable));
    }

    public EncodingSpec(EncodingSpec v1) {
        this.variables.addAll(v1.getVariables());
    }

    public EncodingSpec(EncodingSpec v1, EncodingSpec v2) {
        this.variables.addAll(v1.getVariables());
        this.variables.addAll(v2.getVariables());
        for (Set<String> x1 : v1.getVariables()) {
            for (Set<String> x2 : v2.getVariables()) {
                Set<String> both = Sets.newTreeSet(x1);
                both.addAll(x2);
                variables.add(both);
            }
        }
    }

    private void add(EncodingSpec spec) {
        variables.addAll(spec.getVariables());
    }

    public Set<Set<String>> getVariables() {
        return variables;
    }

    public static EncodingSpec add(EncodingSpec a, EncodingSpec b) {
        EncodingSpec r = new EncodingSpec(a);
        r.add(b);
        return r;
    }

    public static EncodingSpec interact(EncodingSpec a, EncodingSpec b) {
        return new EncodingSpec(a, b);
    }

    public static EncodingSpec pow(EncodingSpec a, int n) {
        EncodingSpec r = new EncodingSpec(a);
        if (n > r.getVariables().size()) {
            n = r.getVariables().size();
        }
        for (int i = 1; i < n; i++) {
            r = new EncodingSpec(r, a);
        }
        return r;
    }

    public static EncodingSpec atom(String s) {
        return new EncodingSpec(s);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import org.antlr.runtime.ANTLRReaderStream;
import org.antlr.runtime.CommonTokenStream;
import org.antlr.runtime.RecognitionException;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;

public class Formula {
    public static EncodingSpec parse(Reader reader) {
        FormulaParser tokenParser;
        try {
            CommonTokenStream input = new CommonTokenStream(new FormulaLexer(new ANTLRReaderStream(reader)));
            tokenParser = new FormulaParser(input);
            FormulaParser.expression_return parserResult = tokenParser.expression();
            return parserResult.r;
        } catch (IOException e) {
            throw new SchemaParseException("Cannot parse schema", e);
        } catch (RecognitionException e) {
            throw new SchemaParseException("Cannot parse schema", e);
        }
    }

    public static EncodingSpec parse(String formula) {
        final StringReader input = new StringReader(formula);
        EncodingSpec r = parse(input);
        input.close();
        return r;
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

public class FormulaParseException extends RuntimeException {
    public FormulaParseException(Throwable e, String msg) {
        super(msg, e);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import org.antlr.runtime.ANTLRReaderStream;
import org.antlr.runtime.CommonTokenStream;
import org.antlr.runtime.RecognitionException;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.util.Map;

/**
 * A schema specifies the names and types for positional parameters in a tuple.  These names
 * and types are used when encoding tuples as vectors using PigVector.
 * <p/>
 * This class contains convenience routines that simplify the use of the schema parser classes.
 * <p/>
 * A schema consists of a comma separated list of variable specifications.  A variable
 * specification contains a name plus optional absolute position and a type separated by a colon.
 * The name follows the rules of java identifiers.  The optional absolute position is indicated
 * by a dollar sign and a number.  The number indicates the 0-based position of the variable
 * in any tuple being encoded.  Variables without positions are assigned consecutive positions
 * starting at 0 in the tuple.
 * <p/>
 * The type specification can be numeric, word or text.  For text, you can supply an optional
 * parenthesis surrounded fully qualified name of a Java class that implements the Analyzer
 * interface from Lucene.  If you don't supply an analyzer, the text is tokenized on whilte-
 * space boundaries.
 * <p/>
 * Here is an example of a schema:
 * <pre>
 * a:numeric, b$0:word, c$1:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text
 * </pre>
 * Here, argument 0 in the encoded tuple is encoded as a number and as a key word while
 * argument 1 in the encoded tuple is encoded as text using two different methods for
 * analysis.
 */
public class Schema {
    public static Map<String, ArgumentEncoder> parse(Reader reader) {
        SchemaParser tokenParser;
        try {
            CommonTokenStream input = new CommonTokenStream(new SchemaLexer(new ANTLRReaderStream(reader)));
            tokenParser = new SchemaParser(input);
            // parsing causes side effect on tokenParser
            tokenParser.schema();
        } catch (IOException e) {
            throw new SchemaParseException("Cannot parse schema", e);
        } catch (RecognitionException e) {
            throw new SchemaParseException("Cannot parse schema", e);
        }
        return tokenParser.getEncoders();
    }

    public static Map<String, ArgumentEncoder> parse(String s) {
        final StringReader input = new StringReader(s);
        Map<String, ArgumentEncoder> r = parse(input);
        input.close();
        return r;
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

public class SchemaParseException extends RuntimeException {
    public SchemaParseException(String s, Throwable throwable) {
        super(s, throwable);
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import org.apache.mahout.math.Vector;
import org.apache.mahout.math.function.Functions;
import org.apache.mahout.pig.encoders.EncodeVector;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.junit.Test;

import java.io.IOException;

import static org.junit.Assert.*;

public class EncodeVectorTest {

    @Test
    public void testSimple() throws IOException {
        EncodeVector ev = new EncodeVector("100", "1+x+y", "x:numeric, not:text, y:text");
        Tuple t = TupleFactory.getInstance().newTuple(4);
        t.set(0, 3.1);
        t.set(1, "foo bar");
        t.set(2, "");
        t.set(3, null);

        Vector v1 = PigVector.fromBytes(ev.exec(t));
        assertEquals(3.1, v1.get(10), 0);
        assertEquals(1.0, v1.get(39), 0);

        t.set(2, "fob");

        Vector pv = PigVector.fromBytes(ev.exec(t));
        Vector v2 = pv.assign(v1, Functions.MINUS);
        assertEquals(1, v2.get(40), 0);
        assertEquals(1, v2.get(55), 0);

        t.set(2, "lots of text for testing");

        pv = PigVector.fromBytes(ev.exec(t));
        Vector v3 = pv.assign(v1, Functions.MINUS);
        assertEquals(10, v3.zSum(), 0);
        assertEquals(1, v3.maxValue(), 1);

        EncodeVector ev2 = new EncodeVector("100", "x+y", "x:numeric, not:text, y:text");
        t.set(2, "");
        pv = PigVector.fromBytes(ev2.exec(t));
        Vector v4 = pv;
        assertEquals(3.1, v1.get(10), 0);
        assertEquals(3.1, v4.zSum(), 0);
        assertEquals(3.1, v4.maxValue(), 1);
    }

    @Test
    public void testBadIndex() throws IOException {
        EncodeVector ev2 = new EncodeVector("100", "x+y", "x:numeric, not:text, y:text");
        Tuple t = TupleFactory.getInstance().newTuple(2);
        t.set(0, 3.1);
        t.set(1, "foo bar");
        try {
            ev2.exec(t);
            fail("Should have failed because tuple is too small");
        } catch (IllegalArgumentException e) {
            assertTrue(e.getMessage().contains("Tuple doesn't have at least"));
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import org.antlr.runtime.ANTLRReaderStream;
import org.antlr.runtime.CommonTokenStream;
import org.antlr.runtime.RecognitionException;
import org.apache.mahout.pig.encoders.EncodingSpec;
import org.apache.mahout.pig.encoders.FormulaLexer;
import org.apache.mahout.pig.encoders.FormulaParseException;
import org.apache.mahout.pig.encoders.FormulaParser;
import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;

import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class EncodingSpecTest {
    @Test
    public void testSum() throws RecognitionException, IOException {
        EncodingSpec z = parse("a+c+b");
        Assert.assertEquals("[[a], [b], [c]]", z.getVariables().toString());

        try {
            parse("a+$1+$2");
            fail("Should have failed");
        } catch (FormulaParseException e) {
            assertTrue(e.getMessage().startsWith("Syntax error"));
        }

        z = parse("a + 1");
        Assert.assertEquals("[[1], [a]]", z.getVariables().toString());
    }

    @Test
    public void testWrongBias() throws RecognitionException, IOException {
        try {
            parse("a + 2");
            fail("Should have failed due to invalid bias");
        } catch (FormulaParseException e) {
            assertTrue(e.getMessage().startsWith("Invalid bias"));
        }
    }

    @Test
    public void testProduct() throws RecognitionException, IOException {
        EncodingSpec z = parse("a+c*b");
        Assert.assertEquals("[[a], [b], [b, c], [c]]", z.getVariables().toString());
        z = parse("(a+c)*(c+b+a)");
        Assert.assertEquals("[[a], [a, b], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
    }

    @Test
    public void testPower() throws RecognitionException, IOException {
        EncodingSpec z = parse("(a+b+c)^2");
        Assert.assertEquals("[[a], [a, b], [a, c], [b], [b, c], [c]]", z.getVariables().toString());

        z = parse("(a+b+c)^3");
        Assert.assertEquals("[[a], [a, b], [a, b, c], [a, c], [b], [b, c], [c]]", z.getVariables().toString());

        z = parse("(a+b+c)^30");
        Assert.assertEquals("[[a], [a, b], [a, b, c], [a, c], [b], [b, c], [c]]", z.getVariables().toString());
    }

    private EncodingSpec parse(String s) throws IOException, RecognitionException {
        Reader reader = new StringReader(s);
        CommonTokenStream input = new CommonTokenStream(new FormulaLexer(new ANTLRReaderStream(reader)));
        FormulaParser tokenParser = new FormulaParser(input);
        FormulaParser.expression_return parserResult = tokenParser.expression();
        reader.close();
        return parserResult.r;
    }

}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import com.google.common.collect.ImmutableList;
import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.function.DoubleFunction;
import org.apache.pig.data.*;
import org.junit.Test;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.List;
import java.util.Random;

import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.fail;
import static org.junit.Assert.assertTrue;

public class LogisticRegressionTest {
    @Test
    public void testConstructor() throws IOException {
        LogisticRegression lr = new LogisticRegression("categories = a b c, features=10000");
        assertEquals(3, lr.getModel().numCategories());
        assertEquals(10000, lr.getModel().numFeatures());
        assertEquals(1, lr.getIterations());
        assertTrue(lr.isInMemory());


        lr = new LogisticRegression("categories = a b c, features=10000, decayExponent=0.3 ,stepOffset=123, learningRate=2.1, lambda=0.001, iterations=12, inMemory=false");
        assertEquals(2.1 * Math.pow(123, -0.3), lr.getModel().currentLearningRate());
        assertEquals(0.001, lr.getModel().getLambda());

        try {
            new LogisticRegression("categories = a , features=10000");
            fail("Should have failed");
        } catch (BadClassifierSpecException e) {
            assertTrue("Single target category", e.getMessage().startsWith("Must have more than one target category"));
        }

        try {
            new LogisticRegression("categories = a b, features=10000, x=3");
            fail("Should have failed");
        } catch (BadClassifierSpecException e) {
            assertEquals("Extra options", "Extra options supplied: x", e.getMessage());
        }

        try {
            new LogisticRegression("categories = a ");
            fail("Should have failed");
        } catch (BadClassifierSpecException e) {
            assertEquals("No features", "Must specify previous model location using \"file\" or supply \"categories\" and \"features\"", e.getMessage());
        }
    }

    @Test
    public void testTraining() throws IOException {
        DoubleFunction randomValue = new DoubleFunction() {
            private Random gen = new Random(1);

            public double apply(double arg1) {
                return gen.nextGaussian();
            }
        };

        // start with a random direction
        Vector n = new DenseVector(4);
        n.assign(randomValue);

        // make up data.  result is 1 if in the same rough direction as n, else 0
        String[] target = new String[]{"0", "1"};
        DataBag examples = new DefaultDataBag();
        for (int i = 0; i < 400; i++) {
            Vector v = new DenseVector(4);
            v.assign(randomValue);

            Tuple x = new DefaultTuple();
            x.append(target[v.dot(n) > 0 ? 1 : 0]);
            x.append(PigVector.toBytes(v));
            examples.add(x);
        }
        Tuple data = new DefaultTuple();
        data.append(examples);

        // train model.  training from tmp file allows absolute repeatability
        LogisticRegression lr = new LogisticRegression("categories = 0 1, features=4, inMemory=false, iterations=5");
        lr.accumulate(data);
        DataByteArray r = lr.getValue();
        DataInputStream in = new DataInputStream(new ByteArrayInputStream(r.get()));
        Classifier c = PolymorphicWritable.read(in, Classifier.class);
        assertEquals(2, c.getCategories().size());
        assertEquals("0", c.getCategories().get(0));
        assertEquals("1", c.getCategories().get(1));
        OnlineLogisticRegression model = c.getModel();
        assertEquals(lr.getModel().currentLearningRate(), model.currentLearningRate(), 1e-10);
        in.close();

        // with that many data points, model should point in the same direction as the original vector
        Vector v = model.getBeta().viewRow(0);
        double z = n.dot(v) / (n.norm(2) * v.norm(2));
        assertEquals(1.0, z, 1e-2);

        // just for grins, we should check whether the model actually computes the correct values
        List<String> categories = ImmutableList.of("0", "1");
        for (Tuple example : examples) {
            double score = model.classifyScalar(PigVector.fromBytes((DataByteArray) example.get(1)));
            int actual = categories.indexOf(example.get(0));
            score = score * actual + (1 - score) * (1 - actual);
            assertTrue(score > 0.4);
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig;

import org.antlr.runtime.RecognitionException;
import org.apache.mahout.pig.encoders.ArgumentEncoder;
import org.apache.mahout.pig.encoders.Schema;
import org.apache.mahout.pig.encoders.SchemaParseException;
import org.apache.mahout.vectorizer.encoders.ContinuousValueEncoder;
import org.apache.mahout.vectorizer.encoders.LuceneTextValueEncoder;
import org.apache.mahout.vectorizer.encoders.StaticWordValueEncoder;
import org.apache.mahout.vectorizer.encoders.TextValueEncoder;
import org.junit.Test;

import java.io.IOException;
import java.util.Map;

import static org.junit.Assert.*;

public class SchemaTest {
    @Test
    public void testBasics() throws RecognitionException, IOException {
        Map<String, ArgumentEncoder> r = Schema.parse("a:numeric, b:word, c:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text");
        assertEquals(4, r.size());
        assertTrue(r.get("a").getEncoder() instanceof ContinuousValueEncoder);
        assertTrue(r.get("b").getEncoder() instanceof StaticWordValueEncoder);
        assertTrue(r.get("c").getEncoder() instanceof LuceneTextValueEncoder);
        assertTrue(r.get("d").getEncoder() instanceof TextValueEncoder);
    }

    @Test
    public void testFancyPositioning() throws RecognitionException, IOException {
        Map<String, ArgumentEncoder> r = Schema.parse("a:numeric, b$0:word, c$1:text(org.apache.lucene.analysis.en.EnglishAnalyzer), d:text");
        assertEquals(4, r.size());
        assertTrue(r.get("a").getEncoder() instanceof ContinuousValueEncoder);
        assertTrue(r.get("b").getEncoder() instanceof StaticWordValueEncoder);
        assertTrue(r.get("c").getEncoder() instanceof LuceneTextValueEncoder);
        assertTrue(r.get("d").getEncoder() instanceof TextValueEncoder);

        assertEquals(0, r.get("a").getPosition());
        assertEquals(0, r.get("b").getPosition());
        assertEquals(1, r.get("c").getPosition());
        assertEquals(1, r.get("d").getPosition());
    }

    @Test
    public void testMultipleDefinition() throws RecognitionException, IOException {
        try {
            Schema.parse("a:numeric, b:word, a:text, d:text");
            fail("Should have thrown syntax error");
        } catch (SchemaParseException e) {
            assertTrue(e.getMessage().contains("multiply defined"));
        }
    }

    @Test
    public void testSyntaxError() throws RecognitionException, IOException {
        try {
            Schema.parse("a, b:word, a:text, d:text");
            fail("Should have thrown syntax error");
        } catch (SchemaParseException e) {
            assertTrue(e.getMessage().contains("Syntax error"));
        }
    }
}
/*
 * Copyright 2014 Ted Dunning
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.mahout.pig.encoders;

import org.apache.mahout.math.Vector;
import org.apache.mahout.pig.PigVector;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.junit.Test;

import java.io.IOException;

import static junit.framework.Assert.*;

public class DirectVectorTest {
    @Test
    public void testEncoding() throws IOException {
        DirectVector dv = new DirectVector("a,b,c");
        final Tuple input = TupleFactory.getInstance().newTuple();
        input.append(1.0);
        input.append(2);
        input.append(3L);
        Vector v = PigVector.fromBytes(dv.exec(input));
        assertEquals(3, v.size());
        assertEquals(1.0, v.get(0), 0);
        assertEquals(2.0, v.get(1), 0);
        assertEquals(3.0, v.get(2), 0);
    }

    @Test
    public void testWrongSize() throws IOException {
        DirectVector dv = new DirectVector("a,b,c");
        final Tuple input = TupleFactory.getInstance().newTuple();
        input.append(1.0);
        input.append(2);
        try {
            dv.exec(input);
            fail("Should have noticed short tuple");
        } catch (IllegalArgumentException e) {
            assertEquals("Tuple doesn't have 3 elements", e.getMessage());
        }
    }


    @Test
    public void testNullAsNaN() throws IOException {
        DirectVector dv = new DirectVector("a,b,c");
        final Tuple input = TupleFactory.getInstance().newTuple();
        input.append(1.0);
        input.append(2);
        input.append(null);

        Vector v = PigVector.fromBytes(dv.exec(input));
        assertEquals(3, v.size());
        assertEquals(1.0, v.get(0), 0);
        assertEquals(2.0, v.get(1), 0);
        assertTrue(Double.isNaN(v.get(2)));
    }
}
register 'target/pig-vector-1.0-jar-with-dependencies.jar';
define EncodeVector org.apache.mahout.pig.encoders.EncodeVector('10','x+y+1', 'x:numeric, y:numeric');
define DirectVector org.apache.mahout.pig.encoders.DirectVector('x1,x2,x3');

a = load '/Users/tdunning/Downloads/NNBench.csv' using PigStorage(',') as (x1:int, x2:int, x3:int);
b = foreach a generate EncodeVector(*), DirectVector(*);
dump b;         
store b into 'foo.bin' using BinStorage;