chore: import upstream snapshot with attribution

datavec/datavec-local/pom.xml

@@ -0,0 +1,90 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ /* ******************************************************************************
+  ~  *
+  ~  *
+  ~  * This program and the accompanying materials are made available under the
+  ~  * terms of the Apache License, Version 2.0 which is available at
+  ~  * https://www.apache.org/licenses/LICENSE-2.0.
+  ~  *
+  ~  *  See the NOTICE file distributed with this work for additional
+  ~  *  information regarding copyright ownership.
+  ~  * 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.
+  ~  *
+  ~  * SPDX-License-Identifier: Apache-2.0
+  ~  ******************************************************************************/
+  -->
+
+<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>
+
+    <parent>
+        <groupId>org.eclipse.deeplearning4j</groupId>
+        <artifactId>datavec-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>datavec-local</artifactId>
+
+    <name>datavec-local</name>
+
+    <properties>
+
+
+        <module.name>datavec.local</module.name>
+    </properties>
+
+
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.moditect</groupId>
+                <artifactId>moditect-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.codepoetics</groupId>
+            <artifactId>protonpack</artifactId>
+            <version>${protonpack.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.deeplearning4j</groupId>
+            <artifactId>datavec-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.deeplearning4j</groupId>
+            <artifactId>datavec-arrow</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.findbugs</groupId>
+                    <artifactId>jsr305</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.deeplearning4j</groupId>
+            <artifactId>nd4j-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.deeplearning4j</groupId>
+            <artifactId>python4j-numpy</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+    </dependencies>
+
+
+</project>

datavec/datavec-local/src/main/java/org/datavec/local/transforms/AnalyzeLocal.java

@@ -0,0 +1,228 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.records.reader.SequenceRecordReader;
+import org.datavec.api.transform.ColumnType;
+import org.datavec.api.transform.analysis.AnalysisCounter;
+import org.datavec.api.transform.analysis.DataAnalysis;
+import org.datavec.api.transform.analysis.DataVecAnalysisUtils;
+import org.datavec.api.transform.analysis.columns.ColumnAnalysis;
+import org.datavec.api.transform.analysis.histogram.HistogramCounter;
+import org.datavec.api.transform.analysis.quality.QualityAnalysisAddFunction;
+import org.datavec.api.transform.analysis.quality.QualityAnalysisState;
+import org.datavec.api.transform.quality.DataQualityAnalysis;
+import org.datavec.api.transform.quality.columns.ColumnQuality;
+import org.datavec.api.transform.schema.Schema;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.analysis.aggregate.AnalysisAddFunction;
+import org.datavec.local.transforms.analysis.histogram.HistogramAddFunction;
+
+import java.util.*;
+
+public class AnalyzeLocal {
+    private static final int DEFAULT_MAX_HISTOGRAM_BUCKETS = 30;
+
+    /**
+     * Analyse the specified data - returns a DataAnalysis object with summary information about each column
+     *
+     * @param schema Schema for data
+     * @param rr     Data to analyze
+     * @return DataAnalysis for data
+     */
+    public static DataAnalysis analyze(Schema schema, RecordReader rr) {
+        return analyze(schema, rr, DEFAULT_MAX_HISTOGRAM_BUCKETS);
+    }
+
+    /**
+     * Analyse the specified data - returns a DataAnalysis object with summary information about each column
+     *
+     * @param schema Schema for data
+     * @param rr     Data to analyze
+     * @return DataAnalysis for data
+     */
+    public static DataAnalysis analyze(Schema schema, RecordReader rr, int maxHistogramBuckets){
+        AnalysisAddFunction addFn = new AnalysisAddFunction(schema);
+        List<AnalysisCounter> counters = null;
+        while(rr.hasNext()){
+            counters = addFn.apply(counters, rr.next());
+        }
+
+        double[][] minsMaxes = new double[counters.size()][2];
+
+        List<ColumnType> columnTypes = schema.getColumnTypes();
+        List<ColumnAnalysis> list = DataVecAnalysisUtils.convertCounters(counters, minsMaxes, columnTypes);
+
+
+        //Do another pass collecting histogram values:
+        List<HistogramCounter> histogramCounters = null;
+        HistogramAddFunction add = new HistogramAddFunction(maxHistogramBuckets, schema, minsMaxes);
+        if(rr.resetSupported()){
+            rr.reset();
+            while(rr.hasNext()){
+                histogramCounters = add.apply(histogramCounters, rr.next());
+            }
+
+            DataVecAnalysisUtils.mergeCounters(list, histogramCounters);
+        }
+
+        return new DataAnalysis(schema, list);
+    }
+
+
+    /**
+     * Analyze the data quality of sequence data - provides a report on missing values, values that don't comply with schema, etc
+     * @param schema Schema for data
+     * @param data   Data to analyze
+     * @return DataQualityAnalysis object
+     */
+    public static DataQualityAnalysis analyzeQualitySequence(Schema schema, SequenceRecordReader data) {
+        int nColumns = schema.numColumns();
+        List<QualityAnalysisState> states = new ArrayList<>();
+        QualityAnalysisAddFunction addFn = new QualityAnalysisAddFunction(schema);
+        while(data.hasNext()){
+            List<List<Writable>> seq = data.sequenceRecord();
+            for(List<Writable> step : seq){
+                states = addFn.apply(states, step);
+            }
+        }
+
+        List<ColumnQuality> list = new ArrayList<>(nColumns);
+
+        for (QualityAnalysisState qualityState : states) {
+            list.add(qualityState.getColumnQuality());
+        }
+        return new DataQualityAnalysis(schema, list);
+    }
+
+    /**
+     * Analyze the data quality of data - provides a report on missing values, values that don't comply with schema, etc
+     * @param schema Schema for data
+     * @param data   Data to analyze
+     * @return DataQualityAnalysis object
+     */
+    public static DataQualityAnalysis analyzeQuality(final Schema schema, final RecordReader data) {
+        int nColumns = schema.numColumns();
+        List<QualityAnalysisState> states = null;
+        QualityAnalysisAddFunction addFn = new QualityAnalysisAddFunction(schema);
+        while(data.hasNext()){
+            states = addFn.apply(states, data.next());
+        }
+
+        List<ColumnQuality> list = new ArrayList<>(nColumns);
+
+        for (QualityAnalysisState qualityState : states) {
+            list.add(qualityState.getColumnQuality());
+        }
+        return new DataQualityAnalysis(schema, list);
+    }
+
+    /**
+     * Get a list of unique values from the specified columns.
+     * For sequence data, use {@link #getUniqueSequence(List, Schema, SequenceRecordReader)}
+     *
+     * @param columnName    Name of the column to get unique values from
+     * @param schema        Data schema
+     * @param data          Data to get unique values from
+     * @return              List of unique values
+     */
+    public static Set<Writable> getUnique(String columnName, Schema schema, RecordReader data) {
+        int colIdx = schema.getIndexOfColumn(columnName);
+        Set<Writable> unique = new HashSet<>();
+        while(data.hasNext()){
+            List<Writable> next = data.next();
+            unique.add(next.get(colIdx));
+        }
+        return unique;
+    }
+
+    /**
+     * Get a list of unique values from the specified columns.
+     * For sequence data, use {@link #getUniqueSequence(String, Schema, SequenceRecordReader)}
+     *
+     * @param columnNames   Names of the column to get unique values from
+     * @param schema        Data schema
+     * @param data          Data to get unique values from
+     * @return              List of unique values, for each of the specified columns
+     */
+    public static Map<String,Set<Writable>> getUnique(List<String> columnNames, Schema schema, RecordReader data){
+        Map<String,Set<Writable>> m = new HashMap<>();
+        for(String s : columnNames){
+            m.put(s, new HashSet<>());
+        }
+        while(data.hasNext()){
+            List<Writable> next = data.next();
+            for(String s : columnNames){
+                int idx = schema.getIndexOfColumn(s);
+                m.get(s).add(next.get(idx));
+            }
+        }
+        return m;
+    }
+
+    /**
+     * Get a list of unique values from the specified column of a sequence
+     *
+     * @param columnName      Name of the column to get unique values from
+     * @param schema          Data schema
+     * @param sequenceData    Sequence data to get unique values from
+     * @return
+     */
+    public static Set<Writable> getUniqueSequence(String columnName, Schema schema,
+                                                   SequenceRecordReader sequenceData) {
+        int colIdx = schema.getIndexOfColumn(columnName);
+        Set<Writable> unique = new HashSet<>();
+        while(sequenceData.hasNext()){
+            List<List<Writable>> next = sequenceData.sequenceRecord();
+            for(List<Writable> step : next){
+                unique.add(step.get(colIdx));
+            }
+        }
+        return unique;
+    }
+
+    /**
+     * Get a list of unique values from the specified columns of a sequence
+     *
+     * @param columnNames     Name of the columns to get unique values from
+     * @param schema          Data schema
+     * @param sequenceData    Sequence data to get unique values from
+     * @return
+     */
+    public static Map<String,Set<Writable>> getUniqueSequence(List<String> columnNames, Schema schema,
+                                                               SequenceRecordReader sequenceData) {
+        Map<String,Set<Writable>> m = new HashMap<>();
+        for(String s : columnNames){
+            m.put(s, new HashSet<>());
+        }
+        while(sequenceData.hasNext()){
+            List<List<Writable>> next = sequenceData.sequenceRecord();
+            for(List<Writable> step : next) {
+                for (String s : columnNames) {
+                    int idx = schema.getIndexOfColumn(s);
+                    m.get(s).add(step.get(idx));
+                }
+            }
+        }
+        return m;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/BaseFlatMapFunctionAdaptee.java

@@ -0,0 +1,44 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+
+import org.datavec.local.transforms.functions.FlatMapFunctionAdapter;
+
+import java.util.List;
+
+public class BaseFlatMapFunctionAdaptee<K, V>  {
+
+    protected final FlatMapFunctionAdapter<K, V> adapter;
+
+    public BaseFlatMapFunctionAdaptee(FlatMapFunctionAdapter<K, V> adapter) {
+        this.adapter = adapter;
+    }
+
+    public List<V> call(K k)  {
+        try {
+            return adapter.call(k);
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/LocalTransformExecutor.java

@@ -0,0 +1,603 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+import com.codepoetics.protonpack.Indexed;
+import com.codepoetics.protonpack.StreamUtils;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.datavec.api.transform.DataAction;
+import org.datavec.api.transform.Transform;
+import org.datavec.api.transform.TransformProcess;
+import org.datavec.api.transform.filter.Filter;
+import org.datavec.api.transform.join.Join;
+import org.datavec.api.transform.ops.IAggregableReduceOp;
+import org.datavec.api.transform.rank.CalculateSortedRank;
+import org.datavec.api.transform.reduce.IAssociativeReducer;
+import org.datavec.api.transform.schema.Schema;
+import org.datavec.api.transform.schema.SequenceSchema;
+import org.datavec.api.transform.sequence.ConvertToSequence;
+import org.datavec.api.transform.sequence.SequenceSplit;
+import org.datavec.api.writable.*;
+import org.datavec.arrow.ArrowConverter;
+import org.datavec.local.transforms.functions.EmptyRecordFunction;
+import org.datavec.local.transforms.join.ExecuteJoinFromCoGroupFlatMapFunction;
+import org.datavec.local.transforms.join.ExtractKeysFunction;
+import org.datavec.local.transforms.misc.ColumnAsKeyPairFunction;
+import org.datavec.local.transforms.rank.UnzipForCalculateSortedRankFunction;
+import org.datavec.local.transforms.reduce.MapToPairForReducerFunction;
+import org.datavec.local.transforms.sequence.*;
+import org.datavec.local.transforms.transform.LocalTransformFunction;
+import org.datavec.local.transforms.transform.SequenceSplitFunction;
+import org.datavec.local.transforms.transform.filter.LocalFilterFunction;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.function.FunctionalUtils;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.*;
+import java.util.function.BiPredicate;
+import java.util.stream.Collectors;
+
+import static java.util.stream.Collectors.toList;
+
+@Slf4j
+public class LocalTransformExecutor {
+    //a boolean jvm argument that when the system property is true
+    //will cause some functions to invoke a try catch block and just log errors
+    //returning empty records
+    public final static String LOG_ERROR_PROPERTY = "org.datavec.spark.transform.logerrors";
+
+    private static BufferAllocator bufferAllocator = new RootAllocator(Long.MAX_VALUE);
+
+    /**
+     * Execute the specified TransformProcess with the given input data<br>
+     * Note: this method can only be used if the TransformProcess returns non-sequence data. For TransformProcesses
+     * that return a sequence, use {@link #executeToSequence(List, TransformProcess)}
+     *
+     * @param inputWritables   Input data to process
+     * @param transformProcess TransformProcess to execute
+     * @return Processed data
+     */
+    public static List<List<Writable>> execute(List<List<Writable>> inputWritables,
+                                               TransformProcess transformProcess) {
+        if (transformProcess.getFinalSchema() instanceof SequenceSchema) {
+            throw new IllegalStateException("Cannot return sequence data with this method");
+
+        }
+
+        List<List<Writable>> filteredSequence = inputWritables.parallelStream()
+                .filter(input -> input.size() == transformProcess.getInitialSchema().numColumns()).collect(toList());
+        if(filteredSequence.size() != inputWritables.size()) {
+            log.warn("Filtered out " + (inputWritables.size() - filteredSequence.size()) + " values");
+        }
+        return execute(filteredSequence, null, transformProcess).getFirst();
+    }
+
+    /**
+     * Execute the specified TransformProcess with the given input data<br>
+     * Note: this method can only be used if the TransformProcess
+     * starts with non-sequential data,
+     * but returns <it>sequence</it>
+     * data (after grouping or converting to a sequence as one of the steps)
+     *
+     * @param inputWritables   Input data to process
+     * @param transformProcess TransformProcess to execute
+     * @return Processed (sequence) data
+     */
+    public static List<List<List<Writable>>> executeToSequence(List<List<Writable>> inputWritables,
+                                                               TransformProcess transformProcess) {
+        if (!(transformProcess.getFinalSchema() instanceof SequenceSchema)) {
+            throw new IllegalStateException("Cannot return non-sequence data with this method");
+        }
+
+        return execute(inputWritables, null, transformProcess).getSecond();
+    }
+
+    /**
+     * Execute the specified TransformProcess with the given <i>sequence</i> input data<br>
+     * Note: this method can only be used if the TransformProcess starts with sequence data, but returns <i>non-sequential</i>
+     * data (after reducing or converting sequential data to individual examples)
+     *
+     * @param inputSequence    Input sequence data to process
+     * @param transformProcess TransformProcess to execute
+     * @return Processed (non-sequential) data
+     */
+    public static List<List<Writable>> executeSequenceToSeparate(List<List<List<Writable>>> inputSequence,
+                                                                 TransformProcess transformProcess) {
+        if (transformProcess.getFinalSchema() instanceof SequenceSchema) {
+            throw new IllegalStateException("Cannot return sequence data with this method");
+        }
+
+        return execute(null, inputSequence, transformProcess).getFirst();
+    }
+
+    /**
+     * Execute the specified TransformProcess with the given <i>sequence</i> input data<br>
+     * Note: this method can only be used if the TransformProcess starts with sequence data, and also returns sequence data
+     *
+     * @param inputSequence    Input sequence data to process
+     * @param transformProcess TransformProcess to execute
+     * @return Processed (non-sequential) data
+     */
+    public static List<List<List<Writable>>> executeSequenceToSequence(List<List<List<Writable>>> inputSequence,
+                                                                       TransformProcess transformProcess) {
+        if (!(transformProcess.getFinalSchema() instanceof SequenceSchema)) {
+            List<List<List<Writable>>> ret = new ArrayList<>(inputSequence.size());
+            for(List<List<Writable>> timeStep : inputSequence) {
+                ret.add(execute(timeStep,null, transformProcess).getFirst());
+            }
+
+            return ret;
+        }
+
+        return execute(null, inputSequence, transformProcess).getSecond();
+    }
+
+
+
+    /**
+     * Convert a string time series to
+     * the proper writable set based on the schema.
+     * Note that this does not use arrow.
+     * This just uses normal writable objects.
+     *
+     * @param stringInput the string input
+     * @param schema the schema to use
+     * @return the converted records
+     */
+    public static List<List<String>> convertWritableInputToString(List<List<Writable>> stringInput,Schema schema) {
+        List<List<String>> ret = new ArrayList<>();
+        List<List<String>> timeStepAdd = new ArrayList<>();
+        for(int j = 0; j < stringInput.size(); j++) {
+            List<Writable> record = stringInput.get(j);
+            List<String> recordAdd = new ArrayList<>();
+            for(int k = 0; k < record.size(); k++) {
+                recordAdd.add(record.get(k).toString());
+            }
+
+            timeStepAdd.add(recordAdd);
+        }
+
+
+        return ret;
+    }
+
+
+    /**
+     * Convert a string time series to
+     * the proper writable set based on the schema.
+     * Note that this does not use arrow.
+     * This just uses normal writable objects.
+     *
+     * @param stringInput the string input
+     * @param schema the schema to use
+     * @return the converted records
+     */
+    public static List<List<Writable>> convertStringInput(List<List<String>> stringInput,Schema schema) {
+        List<List<Writable>> ret = new ArrayList<>();
+        List<List<Writable>> timeStepAdd = new ArrayList<>();
+        for(int j = 0; j < stringInput.size(); j++) {
+            List<String> record = stringInput.get(j);
+            List<Writable> recordAdd = new ArrayList<>();
+            for(int k = 0; k < record.size(); k++) {
+                switch(schema.getType(k)) {
+                    case Double: recordAdd.add(new DoubleWritable(Double.parseDouble(record.get(k)))); break;
+                    case Float:  recordAdd.add(new FloatWritable(Float.parseFloat(record.get(k)))); break;
+                    case Integer:  recordAdd.add(new IntWritable(Integer.parseInt(record.get(k)))); break;
+                    case Long:  recordAdd.add(new LongWritable(Long.parseLong(record.get(k)))); break;
+                    case String: recordAdd.add(new Text(record.get(k))); break;
+                    case Time: recordAdd.add(new LongWritable(Long.parseLong(record.get(k)))); break;
+
+                }
+            }
+
+            timeStepAdd.add(recordAdd);
+        }
+
+
+        return ret;
+    }
+
+
+
+
+    /**
+     * Convert a string time series to
+     * the proper writable set based on the schema.
+     * Note that this does not use arrow.
+     * This just uses normal writable objects.
+     *
+     * @param stringInput the string input
+     * @param schema the schema to use
+     * @return the converted records
+     */
+    public static List<List<List<String>>> convertWritableInputToStringTimeSeries(List<List<List<Writable>>> stringInput,Schema schema) {
+        List<List<List<String>>> ret = new ArrayList<>();
+        for(int i = 0; i < stringInput.size(); i++) {
+            List<List<Writable>> currInput = stringInput.get(i);
+            List<List<String>> timeStepAdd = new ArrayList<>();
+            for(int j = 0; j < currInput.size(); j++) {
+                List<Writable> record = currInput.get(j);
+                List<String> recordAdd = new ArrayList<>();
+                for(int k = 0; k < record.size(); k++) {
+                    recordAdd.add(record.get(k).toString());
+                }
+
+                timeStepAdd.add(recordAdd);
+            }
+
+            ret.add(timeStepAdd);
+        }
+
+        return ret;
+    }
+
+
+    /**
+     * Convert a string time series to
+     * the proper writable set based on the schema.
+     * Note that this does not use arrow.
+     * This just uses normal writable objects.
+     *
+     * @param stringInput the string input
+     * @param schema the schema to use
+     * @return the converted records
+     */
+    public static List<List<List<Writable>>> convertStringInputTimeSeries(List<List<List<String>>> stringInput,Schema schema) {
+        List<List<List<Writable>>> ret = new ArrayList<>();
+        for(int i = 0; i < stringInput.size(); i++) {
+            List<List<String>> currInput = stringInput.get(i);
+            List<List<Writable>> timeStepAdd = new ArrayList<>();
+            for(int j = 0; j < currInput.size(); j++) {
+                List<String> record = currInput.get(j);
+                List<Writable> recordAdd = new ArrayList<>();
+                for(int k = 0; k < record.size(); k++) {
+                    switch(schema.getType(k)) {
+                        case Double: recordAdd.add(new DoubleWritable(Double.parseDouble(record.get(k)))); break;
+                        case Float:  recordAdd.add(new FloatWritable(Float.parseFloat(record.get(k)))); break;
+                        case Integer:  recordAdd.add(new IntWritable(Integer.parseInt(record.get(k)))); break;
+                        case Long:  recordAdd.add(new LongWritable(Long.parseLong(record.get(k)))); break;
+                        case String: recordAdd.add(new Text(record.get(k))); break;
+                        case Time: recordAdd.add(new LongWritable(Long.parseLong(record.get(k)))); break;
+
+                    }
+                }
+
+                timeStepAdd.add(recordAdd);
+            }
+
+            ret.add(timeStepAdd);
+        }
+
+        return ret;
+    }
+
+    /**
+     * Returns true if the executor
+     * is in try catch mode.
+     * @return
+     */
+    public static boolean isTryCatch() {
+        return Boolean.getBoolean(LOG_ERROR_PROPERTY);
+    }
+
+    private static Pair<List<List<Writable>>, List<List<List<Writable>>>> execute(
+            List<List<Writable>> inputWritables, List<List<List<Writable>>> inputSequence,
+            TransformProcess sequence) {
+        List<List<Writable>> currentWritables = inputWritables;
+        List<List<List<Writable>>> currentSequence = inputSequence;
+
+        List<DataAction> dataActions = sequence.getActionList();
+        if (inputWritables != null) {
+            List<Writable> first = inputWritables.get(0);
+            if (first.size() != sequence.getInitialSchema().numColumns()) {
+                throw new IllegalStateException("Input data number of columns (" + first.size()
+                        + ") does not match the number of columns for the transform process ("
+                        + sequence.getInitialSchema().numColumns() + ")");
+            }
+        } else {
+            List<List<Writable>> firstSeq = inputSequence.get(0);
+            if (firstSeq.size() > 0 && firstSeq.get(0).size() != sequence.getInitialSchema().numColumns()) {
+                throw new IllegalStateException("Input sequence data number of columns (" + firstSeq.get(0).size()
+                        + ") does not match the number of columns for the transform process ("
+                        + sequence.getInitialSchema().numColumns() + ")");
+            }
+        }
+
+
+        for (DataAction d : dataActions) {
+            //log.info("Starting execution of stage {} of {}", count, dataActions.size());     //
+
+            if (d.getTransform() != null) {
+                Transform t = d.getTransform();
+                if (currentWritables != null) {
+                    Function<List<Writable>, List<Writable>> function = new LocalTransformFunction(t);
+                    if (isTryCatch())
+                        currentWritables = currentWritables.stream()
+                                .map(input -> function.apply(input)).filter(input -> new EmptyRecordFunction().apply(input)).collect(toList());
+                    else
+                        currentWritables = currentWritables.stream()
+                                .map(input -> function.apply(input)).collect(toList());
+                } else {
+                    Function<List<List<Writable>>, List<List<Writable>>> function =
+                            new LocalSequenceTransformFunction(t);
+                    if (isTryCatch())
+                        currentSequence = currentSequence.stream()
+                                .map(input -> function.apply(input)).filter(input ->
+                                new SequenceEmptyRecordFunction().apply(input)).collect(toList());
+                    else
+                        currentSequence = currentSequence.stream()
+                                .map(input -> function.apply(input)).collect(toList());
+
+
+                }
+            } else if (d.getFilter() != null) {
+                //Filter
+                Filter f = d.getFilter();
+                if (currentWritables != null) {
+                    LocalFilterFunction localFilterFunction = new LocalFilterFunction(f);
+                    currentWritables = currentWritables.stream()
+                            .filter(input -> localFilterFunction.apply(input)).collect(toList());
+                } else {
+                    LocalSequenceFilterFunction localSequenceFilterFunction = new LocalSequenceFilterFunction(f);
+                    currentSequence = currentSequence.stream().filter(input -> localSequenceFilterFunction.apply(input)).collect(toList());
+                }
+
+            } else if (d.getConvertToSequence() != null) {
+                //Convert to a sequence...
+                final ConvertToSequence cts = d.getConvertToSequence();
+
+                if(cts.isSingleStepSequencesMode()) {
+                    ConvertToSequenceLengthOne convertToSequenceLengthOne = new ConvertToSequenceLengthOne();
+                    //Edge case: create a sequence from each example, by treating each value as a sequence of length 1
+                    currentSequence = currentWritables.stream()
+                            .map(input -> convertToSequenceLengthOne.apply(input))
+                            .collect(toList());
+                    currentWritables = null;
+                } else {
+                    //Standard case: join by key
+                    //First: convert to PairRDD
+                    Schema schema = cts.getInputSchema();
+                    int[] colIdxs = schema.getIndexOfColumns(cts.getKeyColumns());
+                    LocalMapToPairByMultipleColumnsFunction localMapToPairByMultipleColumnsFunction = new LocalMapToPairByMultipleColumnsFunction(colIdxs);
+                    List<Pair<List<Writable>, List<Writable>>> withKey =
+                            currentWritables.stream()
+                                    .map(inputSequence2 -> localMapToPairByMultipleColumnsFunction
+                                            .apply(inputSequence2))
+                                    .collect(toList());
+
+
+                    Map<List<Writable>, List<List<Writable>>> collect = FunctionalUtils.groupByKey(withKey);
+                    LocalGroupToSequenceFunction localGroupToSequenceFunction = new LocalGroupToSequenceFunction(cts.getComparator());
+                    //Now: convert to a sequence...
+                    currentSequence = collect.entrySet().stream()
+                            .map(input -> input.getValue())
+                            .map(input -> localGroupToSequenceFunction.apply(input))
+                            .collect(toList());
+
+                    currentWritables = null;
+                }
+            } else if (d.getConvertFromSequence() != null) {
+                //Convert from sequence...
+
+                if (currentSequence == null) {
+                    throw new IllegalStateException(
+                            "Cannot execute ConvertFromSequence operation: current sequence is null");
+                }
+
+                currentWritables = currentSequence.stream()
+                        .flatMap(input -> input.stream())
+                        .collect(toList());
+                currentSequence = null;
+            } else if (d.getSequenceSplit() != null) {
+                SequenceSplit sequenceSplit = d.getSequenceSplit();
+                if (currentSequence == null)
+                    throw new IllegalStateException("Error during execution of SequenceSplit: currentSequence is null");
+                SequenceSplitFunction sequenceSplitFunction = new SequenceSplitFunction(sequenceSplit);
+
+                currentSequence = currentSequence.stream()
+                        .flatMap(input -> sequenceSplitFunction.call(input).stream())
+                        .collect(toList());
+            } else if (d.getReducer() != null) {
+                final IAssociativeReducer reducer = d.getReducer();
+
+                if (currentWritables == null)
+                    throw new IllegalStateException("Error during execution of reduction: current writables are null. "
+                            + "Trying to execute a reduce operation on a sequence?");
+                MapToPairForReducerFunction mapToPairForReducerFunction = new MapToPairForReducerFunction(reducer);
+                List<Pair<String, List<Writable>>> pair =
+                        currentWritables.stream().map(input -> mapToPairForReducerFunction.apply(input))
+                                .collect(toList());
+
+
+                //initial op
+                Map<String, IAggregableReduceOp<List<Writable>, List<Writable>>> resultPerKey = new HashMap<>();
+
+                val groupedByKey = FunctionalUtils.groupByKey(pair);
+                val aggregated = StreamUtils.aggregate(groupedByKey.entrySet()
+                        .stream(), new BiPredicate<Map.Entry<String, List<List<Writable>>>, Map.Entry<String, List<List<Writable>>>>() {
+                    @Override
+                    public boolean test(Map.Entry<String, List<List<Writable>>> stringListEntry, Map.Entry<String, List<List<Writable>>> stringListEntry2) {
+                        return stringListEntry.getKey().equals(stringListEntry2.getKey());
+                    }
+                }).collect(Collectors.toList());
+
+
+                aggregated.stream().forEach((List<Map.Entry<String, List<List<Writable>>>> input) -> {
+                    for(Map.Entry<String, List<List<Writable>>> entry : input) {
+                        if(!resultPerKey.containsKey(entry.getKey())) {
+                            IAggregableReduceOp<List<Writable>, List<Writable>> reducer2 = reducer.aggregableReducer();
+                            resultPerKey.put(entry.getKey(),reducer2);
+                            for(List<Writable> value : entry.getValue()) {
+                                reducer2.accept(value);
+                            }
+
+                        }
+
+                    }
+                });
+
+
+
+
+                currentWritables = resultPerKey.entrySet().stream()
+                        .map(input -> input.getValue().get()).collect(Collectors.toList());
+
+
+
+            } else if (d.getCalculateSortedRank() != null) {
+                CalculateSortedRank csr = d.getCalculateSortedRank();
+
+                if (currentWritables == null) {
+                    throw new IllegalStateException(
+                            "Error during execution of CalculateSortedRank: current writables are null. "
+                                    + "Trying to execute a CalculateSortedRank operation on a sequence? (not currently supported)");
+                }
+
+                Comparator<Writable> comparator = csr.getComparator();
+                String sortColumn = csr.getSortOnColumn();
+                int sortColumnIdx = csr.getInputSchema().getIndexOfColumn(sortColumn);
+                boolean ascending = csr.isAscending();
+                //NOTE: this likely isn't the most efficient implementation.
+                List<Pair<Writable, List<Writable>>> pairRDD =
+                        currentWritables.stream().map(input -> new ColumnAsKeyPairFunction(sortColumnIdx).apply(input))
+                                .collect(toList());
+                pairRDD = pairRDD.stream().sorted(new Comparator<Pair<Writable, List<Writable>>>() {
+                    @Override
+                    public int compare(Pair<Writable, List<Writable>> writableListPair, Pair<Writable, List<Writable>> t1) {
+                        int result = comparator.compare(writableListPair.getFirst(),t1.getFirst());
+                        if(ascending)
+                            return result;
+                        else
+                            return -result;
+                    }
+                }).collect(toList());
+
+                List<Indexed<Pair<Writable, List<Writable>>>> zipped = StreamUtils.zipWithIndex(pairRDD.stream()).collect(toList());
+                currentWritables = zipped.stream().map(input -> new UnzipForCalculateSortedRankFunction()
+                        .apply(Pair.of(input.getValue(),input.getIndex())))
+                        .collect(toList());
+            } else {
+                throw new RuntimeException("Unknown/not implemented action: " + d);
+            }
+
+        }
+
+        //log.info("Completed {} of {} execution steps", count - 1, dataActions.size());       //Lazy execution means this can be printed before anything has actually happened...
+        if(currentSequence != null) {
+            boolean allSameLength = true;
+            Integer length = null;
+            for(List<List<Writable>> record : currentSequence) {
+                if(length == null) {
+                    length = record.size();
+                }
+                else if(record.size() != length)  {
+                    allSameLength = false;
+                }
+            }
+
+            if(allSameLength) {
+                List<FieldVector> arrowColumns = ArrowConverter.toArrowColumnsTimeSeries(
+                        bufferAllocator,
+                        sequence.getFinalSchema(),
+                        currentSequence);
+                 int timeSeriesLength = currentSequence.get(0).size() * currentSequence.get(0).get(0).size();
+              /*   if(currentSequence.get(0).get(0).size() == 1) {
+                     timeSeriesLength = 1;
+                 }*/
+                List<List<List<Writable>>> writablesConvert = ArrowConverter.toArrowWritablesTimeSeries(
+                        arrowColumns,
+                        sequence.getFinalSchema(),
+                        timeSeriesLength);
+                currentSequence = writablesConvert;
+            }
+
+
+            return Pair.of(null, currentSequence);
+
+        }
+        else {
+
+            return new Pair<>(ArrowConverter.
+                    toArrowWritables(ArrowConverter.toArrowColumns(
+                            bufferAllocator,
+                            sequence.getFinalSchema(),
+                            currentWritables)
+                            ,sequence.getFinalSchema()),
+                    null);
+        }
+
+    }
+
+
+
+
+    /**
+     * Execute a join on the specified data
+     *
+     * @param join  Join to execute
+     * @param left  Left data for join
+     * @param right Right data for join
+     * @return Joined data
+     */
+    public static List<List<Writable>> executeJoin(Join join, List<List<Writable>> left,
+                                                   List<List<Writable>> right) {
+
+        String[] leftColumnNames = join.getJoinColumnsLeft();
+        int[] leftColumnIndexes = new int[leftColumnNames.length];
+        for (int i = 0; i < leftColumnNames.length; i++) {
+            leftColumnIndexes[i] = join.getLeftSchema().getIndexOfColumn(leftColumnNames[i]);
+        }
+        ExtractKeysFunction extractKeysFunction1 = new ExtractKeysFunction(leftColumnIndexes);
+
+        List<Pair<List<Writable>, List<Writable>>> leftJV = left.stream()
+                .filter(input -> input.size() != leftColumnNames.length).map(input ->
+                extractKeysFunction1.apply(input)).collect(toList());
+
+        String[] rightColumnNames = join.getJoinColumnsRight();
+        int[] rightColumnIndexes = new int[rightColumnNames.length];
+        for (int i = 0; i < rightColumnNames.length; i++) {
+            rightColumnIndexes[i] = join.getRightSchema().getIndexOfColumn(rightColumnNames[i]);
+        }
+
+        ExtractKeysFunction extractKeysFunction = new ExtractKeysFunction(rightColumnIndexes);
+        List<Pair<List<Writable>, List<Writable>>> rightJV =
+                right.stream().filter(input -> input.size() != rightColumnNames.length)
+                        .map(input -> extractKeysFunction.apply(input))
+                        .collect(toList());
+
+        Map<List<Writable>, Pair<List<List<Writable>>, List<List<Writable>>>> cogroupedJV = FunctionalUtils.cogroup(leftJV, rightJV);
+        ExecuteJoinFromCoGroupFlatMapFunction executeJoinFromCoGroupFlatMapFunction = new ExecuteJoinFromCoGroupFlatMapFunction(join);
+        List<List<Writable>> ret =  cogroupedJV.entrySet().stream()
+                .flatMap(input ->
+                        executeJoinFromCoGroupFlatMapFunction.call(Pair.of(input.getKey(),input.getValue())).stream())
+                .collect(toList());
+
+        Schema retSchema = join.getOutputSchema();
+        return ArrowConverter.toArrowWritables(ArrowConverter.toArrowColumns(bufferAllocator,retSchema,ret),retSchema);
+
+    }
+
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/LocalTransformProcessRecordReader.java

@@ -0,0 +1,38 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.records.reader.impl.transform.TransformProcessRecordReader;
+import org.datavec.api.transform.TransformProcess;
+
+public class LocalTransformProcessRecordReader extends TransformProcessRecordReader {
+
+    /**
+     * Initialize with the internal record reader
+     * and the transform process.
+     * @param recordReader
+     * @param transformProcess
+     */
+    public LocalTransformProcessRecordReader(RecordReader recordReader, TransformProcess transformProcess) {
+        super(recordReader, transformProcess);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/LocalTransformProcessSequenceRecordReader.java

@@ -0,0 +1,52 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+import org.datavec.api.records.reader.SequenceRecordReader;
+import org.datavec.api.records.reader.impl.transform.TransformProcessSequenceRecordReader;
+import org.datavec.api.transform.TransformProcess;
+import org.datavec.api.writable.Writable;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class LocalTransformProcessSequenceRecordReader extends TransformProcessSequenceRecordReader {
+
+    public LocalTransformProcessSequenceRecordReader(SequenceRecordReader sequenceRecordReader, TransformProcess transformProcess) {
+        super(sequenceRecordReader, transformProcess);
+    }
+
+    @Override
+    public List<List<Writable>> sequenceRecord() {
+        return LocalTransformExecutor.executeSequenceToSequence(Arrays.asList(sequenceRecordReader.nextSequence().getSequenceRecord()),transformProcess
+        ).get(0);
+    }
+
+    @Override
+    public List<List<Writable>> next(int num) {
+        return super.next(num);
+    }
+
+    @Override
+    public List<Writable> next() {
+        return super.next();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/SequenceEmptyRecordFunction.java

@@ -0,0 +1,33 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms;
+
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+public class SequenceEmptyRecordFunction implements Function<List<List<Writable>>, Boolean> {
+    @Override
+    public Boolean apply(List<List<Writable>> v1) {
+        return v1.isEmpty();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/analysis/aggregate/AnalysisAddFunction.java

@@ -0,0 +1,85 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.analysis.aggregate;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.ColumnType;
+import org.datavec.api.transform.analysis.AnalysisCounter;
+import org.datavec.api.transform.analysis.counter.*;
+import org.datavec.api.transform.schema.Schema;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.BiFunction;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@AllArgsConstructor
+public class AnalysisAddFunction implements BiFunction<List<AnalysisCounter>, List<Writable>, List<AnalysisCounter>> {
+    private Schema schema;
+
+    @Override
+    public List<AnalysisCounter> apply(List<AnalysisCounter> analysisCounters, List<Writable> writables){
+        if (analysisCounters == null) {
+            analysisCounters = new ArrayList<>();
+            List<ColumnType> columnTypes = schema.getColumnTypes();
+            for (ColumnType ct : columnTypes) {
+                switch (ct) {
+                    case String:
+                        analysisCounters.add(new StringAnalysisCounter());
+                        break;
+                    case Integer:
+                        analysisCounters.add(new IntegerAnalysisCounter());
+                        break;
+                    case Long:
+                        analysisCounters.add(new LongAnalysisCounter());
+                        break;
+                    case Double:
+                        analysisCounters.add(new DoubleAnalysisCounter());
+                        break;
+                    case Categorical:
+                        analysisCounters.add(new CategoricalAnalysisCounter());
+                        break;
+                    case Time:
+                        analysisCounters.add(new LongAnalysisCounter());
+                        break;
+                    case Bytes:
+                        analysisCounters.add(new BytesAnalysisCounter());
+                        break;
+                    case NDArray:
+                        analysisCounters.add(new NDArrayAnalysisCounter());
+                        break;
+                    default:
+                        throw new IllegalArgumentException("Unknown column type: " + ct);
+                }
+            }
+        }
+
+        int size = analysisCounters.size();
+        if (size != writables.size())
+            throw new IllegalStateException("Writables list and number of counters does not match (" + writables.size()
+                            + " vs " + size + ")");
+        for (int i = 0; i < size; i++) {
+            analysisCounters.get(i).add(writables.get(i));
+        }
+
+        return analysisCounters;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/analysis/aggregate/AnalysisCombineFunction.java

@@ -0,0 +1,48 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.analysis.aggregate;
+
+import org.datavec.api.transform.analysis.AnalysisCounter;
+import org.nd4j.common.function.BiFunction;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class AnalysisCombineFunction
+                implements BiFunction<List<AnalysisCounter>, List<AnalysisCounter>, List<AnalysisCounter>> {
+    @Override
+    public List<AnalysisCounter> apply(List<AnalysisCounter> l1, List<AnalysisCounter> l2) {
+        if (l1 == null)
+            return l2;
+        if (l2 == null)
+            return l1;
+
+        int size = l1.size();
+        if (size != l2.size())
+            throw new IllegalStateException("List lengths differ");
+
+        List<AnalysisCounter> out = new ArrayList<>();
+        for (int i = 0; i < size; i++) {
+            out.add(l1.get(i).merge(l2.get(i)));
+        }
+        return out;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/analysis/histogram/HistogramAddFunction.java

@@ -0,0 +1,94 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.analysis.histogram;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.ColumnType;
+import org.datavec.api.transform.analysis.histogram.*;
+import org.datavec.api.transform.metadata.CategoricalMetaData;
+import org.datavec.api.transform.schema.Schema;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.BiFunction;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@AllArgsConstructor
+public class HistogramAddFunction implements BiFunction<List<HistogramCounter>, List<Writable>, List<HistogramCounter>> {
+    private final int nBins;
+    private final Schema schema;
+    private final double[][] minsMaxes;
+
+    @Override
+    public List<HistogramCounter> apply(List<HistogramCounter> histogramCounters, List<Writable> writables) {
+        if (histogramCounters == null) {
+            histogramCounters = new ArrayList<>();
+            List<ColumnType> columnTypes = schema.getColumnTypes();
+            int i = 0;
+            for (ColumnType ct : columnTypes) {
+                switch (ct) {
+                    case String:
+                        histogramCounters.add(new StringHistogramCounter((int) minsMaxes[i][0], (int) minsMaxes[i][1],
+                                        nBins));
+                        break;
+                    case Integer:
+                        histogramCounters.add(new DoubleHistogramCounter(minsMaxes[i][0], minsMaxes[i][1], nBins));
+                        break;
+                    case Long:
+                        histogramCounters.add(new DoubleHistogramCounter(minsMaxes[i][0], minsMaxes[i][1], nBins));
+                        break;
+                    case Double:
+                        histogramCounters.add(new DoubleHistogramCounter(minsMaxes[i][0], minsMaxes[i][1], nBins));
+                        break;
+                    case Categorical:
+                        CategoricalMetaData meta = (CategoricalMetaData) schema.getMetaData(i);
+                        histogramCounters.add(new CategoricalHistogramCounter(meta.getStateNames()));
+                        break;
+                    case Time:
+                        histogramCounters.add(new DoubleHistogramCounter(minsMaxes[i][0], minsMaxes[i][1], nBins));
+                        break;
+                    case Bytes:
+                        histogramCounters.add(null); //TODO
+                        break;
+                    case NDArray:
+                        histogramCounters.add(new NDArrayHistogramCounter(minsMaxes[i][0], minsMaxes[i][1], nBins));
+                        break;
+                    default:
+                        throw new IllegalArgumentException("Unknown column type: " + ct);
+                }
+
+                i++;
+            }
+        }
+
+        int size = histogramCounters.size();
+        if (size != writables.size())
+            throw new IllegalStateException("Writables list and number of counters does not match (" + writables.size()
+                            + " vs " + size + ")");
+        for (int i = 0; i < size; i++) {
+            HistogramCounter hc = histogramCounters.get(i);
+            if (hc != null)
+                hc.add(writables.get(i));
+        }
+
+        return histogramCounters;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/analysis/histogram/HistogramCombineFunction.java

@@ -0,0 +1,58 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.analysis.histogram;
+
+import org.datavec.api.transform.analysis.histogram.HistogramCounter;
+import org.nd4j.common.function.BiFunction;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HistogramCombineFunction
+                implements BiFunction<List<HistogramCounter>, List<HistogramCounter>, List<HistogramCounter>> {
+    @Override
+    public List<HistogramCounter> apply(List<HistogramCounter> l1, List<HistogramCounter> l2) {
+        if (l1 == null)
+            return l2;
+        if (l2 == null)
+            return l1;
+
+        int size = l1.size();
+        if (size != l2.size())
+            throw new IllegalStateException("List lengths differ");
+
+        List<HistogramCounter> out = new ArrayList<>();
+        for (int i = 0; i < size; i++) {
+            HistogramCounter c1 = l1.get(i);
+            HistogramCounter c2 = l2.get(i);
+
+            //Normally shouldn't get null values here - but maybe for Bytes column, etc.
+            if (c1 == null) {
+                out.add(c2);
+            } else if (c2 == null) {
+                out.add(c1);
+            } else {
+                out.add(c1.merge(c2));
+            }
+        }
+        return out;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/EmptyRecordFunction.java

@@ -0,0 +1,33 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions;
+
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+public class EmptyRecordFunction implements Function<List<Writable>, Boolean> {
+    @Override
+    public Boolean apply(List<Writable> v1) {
+        return v1.isEmpty();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/FlatMapFunctionAdapter.java

@@ -0,0 +1,28 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions;
+
+import java.io.Serializable;
+import java.util.List;
+
+public interface FlatMapFunctionAdapter<T, R> extends Serializable {
+    List<R> call(T t) throws Exception;
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/LineRecordReaderFunction.java

@@ -0,0 +1,46 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions;
+
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.split.StringSplit;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+public class LineRecordReaderFunction implements Function<String, List<Writable>> {
+    private final RecordReader recordReader;
+
+    public LineRecordReaderFunction(RecordReader recordReader) {
+        this.recordReader = recordReader;
+    }
+
+    @Override
+    public List<Writable> apply(String s) {
+        try {
+            recordReader.initialize(new StringSplit(s));
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+        return recordReader.next();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/RecordReaderFunction.java

@@ -0,0 +1,53 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions;
+
+
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.util.List;
+
+public class RecordReaderFunction implements Function<Pair<String, InputStream>, List<Writable>> {
+    protected RecordReader recordReader;
+
+    public RecordReaderFunction(RecordReader recordReader) {
+        this.recordReader = recordReader;
+    }
+
+    @Override
+    public List<Writable> apply(Pair<String, InputStream> value) {
+        URI uri = URI.create(value.getFirst());
+        InputStream ds = value.getRight();
+        try (DataInputStream dis = (DataInputStream) ds) {
+            return recordReader.record(uri, dis);
+        } catch (IOException e) {
+            throw new IllegalStateException("Something went wrong reading file");
+        }
+
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/SequenceRecordReaderFunction.java

@@ -0,0 +1,56 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions;
+
+
+import lombok.extern.slf4j.Slf4j;
+import org.datavec.api.records.reader.SequenceRecordReader;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.util.List;
+
+@Slf4j
+public class SequenceRecordReaderFunction
+                implements Function<Pair<String, InputStream>, List<List<Writable>>> {
+    protected SequenceRecordReader sequenceRecordReader;
+
+    public SequenceRecordReaderFunction(SequenceRecordReader sequenceRecordReader) {
+        this.sequenceRecordReader = sequenceRecordReader;
+    }
+
+    @Override
+    public List<List<Writable>> apply(Pair<String, InputStream> value) {
+        URI uri = URI.create(value.getFirst());
+        try (DataInputStream dis = (DataInputStream) value.getRight()) {
+            return sequenceRecordReader.sequenceRecord(uri, dis);
+        } catch (IOException e) {
+            log.error("",e);
+        }
+
+        throw new IllegalStateException("Something went wrong");
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/data/FilesAsBytesFunction.java

@@ -0,0 +1,44 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions.data;
+
+
+import org.apache.commons.io.IOUtils;
+import org.datavec.api.writable.BytesWritable;
+import org.datavec.api.writable.Text;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+public class FilesAsBytesFunction implements Function<Pair<String, InputStream>, Pair<Text, BytesWritable>> {
+    @Override
+    public Pair<Text, BytesWritable> apply(Pair<String, InputStream> in) {
+        try {
+            return Pair.of(new Text(in.getFirst()), new BytesWritable(IOUtils.toByteArray(in.getSecond())));
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+
+        }
+
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/data/RecordReaderBytesFunction.java

@@ -0,0 +1,57 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions.data;
+
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.writable.BytesWritable;
+import org.datavec.api.writable.Text;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+
+public class RecordReaderBytesFunction implements Function<Pair<Text, BytesWritable>, List<Writable>> {
+
+    private final RecordReader recordReader;
+
+    public RecordReaderBytesFunction(RecordReader recordReader) {
+        this.recordReader = recordReader;
+    }
+
+    @Override
+    public List<Writable> apply(Pair<Text, BytesWritable> v1) {
+        URI uri = URI.create(v1.getRight().toString());
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(v1.getRight().getContent()));
+        try {
+            return recordReader.record(uri, dis);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+
+    }
+
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/functions/data/SequenceRecordReaderBytesFunction.java

@@ -0,0 +1,56 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.functions.data;
+
+
+import org.datavec.api.records.reader.SequenceRecordReader;
+import org.datavec.api.writable.BytesWritable;
+import org.datavec.api.writable.Text;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+
+public class SequenceRecordReaderBytesFunction implements Function<Pair<Text, BytesWritable>, List<List<Writable>>> {
+
+    private final SequenceRecordReader recordReader;
+
+    public SequenceRecordReaderBytesFunction(SequenceRecordReader recordReader) {
+        this.recordReader = recordReader;
+    }
+
+    @Override
+    public List<List<Writable>> apply(Pair<Text, BytesWritable> v1) {
+        URI uri = URI.create(v1.getFirst().toString());
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(v1.getRight().getContent()));
+        try {
+            return recordReader.sequenceRecord(uri, dis);
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/ExecuteJoinFromCoGroupFlatMapFunction.java

@@ -0,0 +1,36 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import org.datavec.api.transform.join.Join;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.BaseFlatMapFunctionAdaptee;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.List;
+
+public class ExecuteJoinFromCoGroupFlatMapFunction extends
+        BaseFlatMapFunctionAdaptee<Pair<List<Writable>, Pair<List<List<Writable>>, List<List<Writable>>>>, List<Writable>> {
+
+    public ExecuteJoinFromCoGroupFlatMapFunction(Join join) {
+        super(new ExecuteJoinFromCoGroupFlatMapFunctionAdapter(join));
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/ExecuteJoinFromCoGroupFlatMapFunctionAdapter.java

@@ -0,0 +1,118 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import org.nd4j.shade.guava.collect.Iterables;
+import org.datavec.api.transform.join.Join;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.functions.FlatMapFunctionAdapter;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ExecuteJoinFromCoGroupFlatMapFunctionAdapter implements
+        FlatMapFunctionAdapter<Pair<List<Writable>, Pair<List<List<Writable>>, List<List<Writable>>>>, List<Writable>> {
+
+    private final Join join;
+
+    public ExecuteJoinFromCoGroupFlatMapFunctionAdapter(Join join) {
+        this.join = join;
+    }
+
+    @Override
+    public List<List<Writable>> call(
+                    Pair<List<Writable>, Pair<List<List<Writable>>, List<List<Writable>>>> t2)
+                    throws Exception {
+
+        Iterable<List<Writable>> leftList = t2.getSecond().getFirst();
+        Iterable<List<Writable>> rightList = t2.getSecond().getSecond();
+
+        List<List<Writable>> ret = new ArrayList<>();
+        Join.JoinType jt = join.getJoinType();
+        switch (jt) {
+            case Inner:
+                //Return records where key columns appear in BOTH
+                //So if no values from left OR right: no return values
+                for (List<Writable> jvl : leftList) {
+                    for (List<Writable> jvr : rightList) {
+                        List<Writable> joined = join.joinExamples(jvl, jvr);
+                        ret.add(joined);
+                    }
+                }
+                break;
+            case LeftOuter:
+                //Return all records from left, even if no corresponding right value (NullWritable in that case)
+                for (List<Writable> jvl : leftList) {
+                    if (Iterables.size(rightList) == 0) {
+                        List<Writable> joined = join.joinExamples(jvl, null);
+                        ret.add(joined);
+                    } else {
+                        for (List<Writable> jvr : rightList) {
+                            List<Writable> joined = join.joinExamples(jvl, jvr);
+                            ret.add(joined);
+                        }
+                    }
+                }
+                break;
+            case RightOuter:
+                //Return all records from right, even if no corresponding left value (NullWritable in that case)
+                for (List<Writable> jvr : rightList) {
+                    if (Iterables.size(leftList) == 0) {
+                        List<Writable> joined = join.joinExamples(null, jvr);
+                        ret.add(joined);
+                    } else {
+                        for (List<Writable> jvl : leftList) {
+                            List<Writable> joined = join.joinExamples(jvl, jvr);
+                            ret.add(joined);
+                        }
+                    }
+                }
+                break;
+            case FullOuter:
+                //Return all records, even if no corresponding left/right value (NullWritable in that case)
+                if (Iterables.size(leftList) == 0) {
+                    //Only right values
+                    for (List<Writable> jvr : rightList) {
+                        List<Writable> joined = join.joinExamples(null, jvr);
+                        ret.add(joined);
+                    }
+                } else if (Iterables.size(rightList) == 0) {
+                    //Only left values
+                    for (List<Writable> jvl : leftList) {
+                        List<Writable> joined = join.joinExamples(jvl, null);
+                        ret.add(joined);
+                    }
+                } else {
+                    //Records from both left and right
+                    for (List<Writable> jvl : leftList) {
+                        for (List<Writable> jvr : rightList) {
+                            List<Writable> joined = join.joinExamples(jvl, jvr);
+                            ret.add(joined);
+                        }
+                    }
+                }
+                break;
+        }
+
+        return ret;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/ExtractKeysFunction.java

@@ -0,0 +1,51 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+@AllArgsConstructor
+public class ExtractKeysFunction implements Function<List<Writable>, Pair<List<Writable>, List<Writable>>> {
+    private int[] columnIndexes;
+
+    @Override
+    public Pair<List<Writable>, List<Writable>> apply(List<Writable> writables) {
+
+        List<Writable> keyValues;
+        if (columnIndexes.length == 1) {
+            keyValues = Collections.singletonList(writables.get(columnIndexes[0]));
+        } else {
+            keyValues = new ArrayList<>(columnIndexes.length);
+            for (int i : columnIndexes) {
+                keyValues.add(writables.get(i));
+            }
+        }
+
+        return Pair.of(keyValues, writables);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/FilterAndFlattenJoinedValues.java

@@ -0,0 +1,35 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import org.datavec.api.transform.join.Join;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.BaseFlatMapFunctionAdaptee;
+
+import java.util.List;
+
+public class FilterAndFlattenJoinedValues extends BaseFlatMapFunctionAdaptee<JoinedValue, List<Writable>> {
+
+    public FilterAndFlattenJoinedValues(Join.JoinType joinType) {
+        super(new FilterAndFlattenJoinedValuesAdapter(joinType));
+    }
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/FilterAndFlattenJoinedValuesAdapter.java

@@ -0,0 +1,68 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import org.datavec.api.transform.join.Join;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.functions.FlatMapFunctionAdapter;
+
+import java.util.Collections;
+import java.util.List;
+
+public class FilterAndFlattenJoinedValuesAdapter implements FlatMapFunctionAdapter<JoinedValue, List<Writable>> {
+
+    private final Join.JoinType joinType;
+
+    public FilterAndFlattenJoinedValuesAdapter(Join.JoinType joinType) {
+        this.joinType = joinType;
+    }
+
+    @Override
+    public List<List<Writable>> call(JoinedValue joinedValue) throws Exception {
+        boolean keep;
+        switch (joinType) {
+            case Inner:
+                //Only keep joined values where we have both left and right
+                keep = joinedValue.isHaveLeft() && joinedValue.isHaveRight();
+                break;
+            case LeftOuter:
+                //Keep all values where left is not missing/null
+                keep = joinedValue.isHaveLeft();
+                break;
+            case RightOuter:
+                //Keep all values where right is not missing/null
+                keep = joinedValue.isHaveRight();
+                break;
+            case FullOuter:
+                //Keep all values
+                keep = true;
+                break;
+            default:
+                throw new RuntimeException("Unknown/not implemented join type: " + joinType);
+        }
+
+        if (keep) {
+            return Collections.singletonList(joinedValue.getValues());
+        } else {
+            return Collections.emptyList();
+        }
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/join/JoinedValue.java

@@ -0,0 +1,38 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.join;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.datavec.api.writable.Writable;
+
+import java.io.Serializable;
+import java.util.List;
+
+@AllArgsConstructor
+@Data
+public class JoinedValue implements Serializable {
+
+    private final boolean haveLeft;
+    private final boolean haveRight;
+    private final List<Writable> values;
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/ColumnAsKeyPairFunction.java

@@ -0,0 +1,38 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class ColumnAsKeyPairFunction implements Function<List<Writable>, Pair<Writable, List<Writable>>> {
+    private final int columnIdx;
+
+    @Override
+    public Pair<Writable, List<Writable>> apply(List<Writable> writables) {
+        return Pair.of(writables.get(columnIdx), writables);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/ColumnToKeyPairTransform.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class ColumnToKeyPairTransform implements Function<List<Writable>, Pair<Writable, Long>> {
+    private final int columnIndex;
+
+    @Override
+    public Pair<Writable, Long> apply(List<Writable> list) {
+        return Pair.of(list.get(columnIndex), 1L);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/NDArrayToWritablesFunction.java

@@ -0,0 +1,55 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.DoubleWritable;
+import org.datavec.api.writable.NDArrayWritable;
+import org.datavec.api.writable.Writable;
+import org.nd4j.linalg.api.ndarray.INDArray;
+import org.nd4j.common.function.Function;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@AllArgsConstructor
+public class NDArrayToWritablesFunction implements Function<INDArray, List<Writable>> {
+    private boolean useNdarrayWritable = false;
+
+    public NDArrayToWritablesFunction() {
+        useNdarrayWritable = false;
+    }
+
+    @Override
+    public List<Writable> apply(INDArray arr) {
+        if (arr.rows() != 1)
+            throw new UnsupportedOperationException("Only NDArray row vectors can be converted to list"
+                                                + " of Writables (found " + arr.rows() + " rows)");
+        List<Writable> record = new ArrayList<>();
+        if (useNdarrayWritable) {
+            record.add(new NDArrayWritable(arr));
+        } else {
+            for (int i = 0; i < arr.columns(); i++)
+                record.add(new DoubleWritable(arr.getDouble(i)));
+        }
+        return record;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/SequenceMergeFunction.java

@@ -0,0 +1,49 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import org.datavec.api.transform.sequence.merge.SequenceMerge;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SequenceMergeFunction<T>
+                implements Function<Pair<T, Iterable<List<List<Writable>>>>, List<List<Writable>>> {
+
+    private SequenceMerge sequenceMerge;
+
+    public SequenceMergeFunction(SequenceMerge sequenceMerge) {
+        this.sequenceMerge = sequenceMerge;
+    }
+
+    @Override
+    public List<List<Writable>> apply(Pair<T, Iterable<List<List<Writable>>>> t2) {
+        List<List<List<Writable>>> sequences = new ArrayList<>();
+        for (List<List<Writable>> l : t2.getSecond()) {
+            sequences.add(l);
+        }
+
+        return sequenceMerge.mergeSequences(sequences);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/SequenceWritablesToStringFunction.java

@@ -0,0 +1,78 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class SequenceWritablesToStringFunction implements Function<List<List<Writable>>, String> {
+
+    public static final String DEFAULT_DELIMITER = ",";
+    public static final String DEFAULT_TIME_STEP_DELIMITER = "\n";
+
+    private final String delimiter;
+    private final String timeStepDelimiter;
+    private final String quote;
+
+    /**
+     * Function with default delimiters ("," and "\n")
+     */
+    public SequenceWritablesToStringFunction() {
+        this(DEFAULT_DELIMITER);
+    }
+
+    /**
+     * function with default delimiter ("\n") between time steps
+     * @param delim Delimiter between values within a single time step
+     */
+    public SequenceWritablesToStringFunction(String delim) {
+        this(delim, DEFAULT_TIME_STEP_DELIMITER, null);
+    }
+
+    /**
+     *
+     * @param delim             The delimiter between column values in a single time step (for example, ",")
+     * @param timeStepDelimiter The delimiter between time steps (for example, "\n")
+     */
+    public SequenceWritablesToStringFunction(String delim, String timeStepDelimiter) {
+        this(delim, timeStepDelimiter, null);
+    }
+
+    @Override
+    public String apply(List<List<Writable>> c) {
+
+        StringBuilder sb = new StringBuilder();
+        boolean firstLine = true;
+        for (List<Writable> timeStep : c) {
+            if (!firstLine) {
+                sb.append(timeStepDelimiter);
+            }
+            WritablesToStringFunction.append(timeStep, sb, delimiter, quote);
+            firstLine = false;
+        }
+
+        return sb.toString();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/StringToWritablesFunction.java

@@ -0,0 +1,54 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.records.reader.RecordReader;
+import org.datavec.api.split.StringSplit;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+@AllArgsConstructor
+public class StringToWritablesFunction implements Function<String, List<Writable>> {
+
+    private RecordReader recordReader;
+
+    @Override
+    public List<Writable> apply(String s) {
+        try {
+            recordReader.initialize(new StringSplit(s));
+        } catch (IOException e) {
+           throw new RuntimeException(e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        }
+        Collection<Writable> next = recordReader.next();
+        if (next instanceof List)
+            return (List<Writable>) next;
+        return new ArrayList<>(next);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/SumLongsFunction2.java

@@ -0,0 +1,32 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+public class SumLongsFunction2 implements Function<Pair<Long, Long>, Long> {
+    @Override
+    public Long apply(Pair<Long, Long> input) {
+        return input.getFirst() + input.getSecond();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/WritablesToNDArrayFunction.java

@@ -0,0 +1,68 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import org.datavec.api.writable.NDArrayWritable;
+import org.datavec.api.writable.Writable;
+import org.nd4j.linalg.api.ndarray.INDArray;
+import org.nd4j.linalg.factory.Nd4j;
+import org.nd4j.common.function.Function;
+import org.nd4j.linalg.indexing.NDArrayIndex;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class WritablesToNDArrayFunction implements Function<List<Writable>, INDArray> {
+
+    @Override
+    public INDArray apply(List<Writable> c) {
+        int length = 0;
+        for (Writable w : c) {
+            if (w instanceof NDArrayWritable) {
+                INDArray a = ((NDArrayWritable) w).get();
+                if (a.isRowVector()) {
+                    length += a.columns();
+                } else {
+                    throw new UnsupportedOperationException("NDArrayWritable is not a row vector."
+                                    + " Can only concat row vectors with other writables. Shape: "
+                                    + Arrays.toString(a.shape()));
+                }
+            } else {
+                length++;
+            }
+        }
+
+        INDArray arr = Nd4j.zeros(1, length);
+        int idx = 0;
+        for (Writable w : c) {
+            if (w instanceof NDArrayWritable) {
+                INDArray subArr = ((NDArrayWritable) w).get();
+                int subLength = subArr.columns();
+                arr.get(NDArrayIndex.point(0), NDArrayIndex.interval(idx, idx + subLength)).assign(subArr);
+                idx += subLength;
+            } else {
+                arr.putScalar(idx++, w.toDouble());
+            }
+        }
+
+        return arr;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/WritablesToStringFunction.java

@@ -0,0 +1,66 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class WritablesToStringFunction implements Function<List<Writable>, String> {
+
+    private final String delim;
+    private final String quote;
+
+    public WritablesToStringFunction(String delim) {
+        this(delim, null);
+    }
+
+    @Override
+    public String apply(List<Writable> c) {
+
+        StringBuilder sb = new StringBuilder();
+        append(c, sb, delim, quote);
+
+        return sb.toString();
+    }
+
+    public static void append(List<Writable> c, StringBuilder sb, String delim, String quote) {
+        boolean first = true;
+        for (Writable w : c) {
+            if (!first)
+                sb.append(delim);
+            String s = w.toString();
+            boolean needQuotes = s.contains(delim);
+            if (needQuotes && quote != null) {
+                sb.append(quote);
+                s = s.replace(quote, quote + quote);
+            }
+            sb.append(s);
+            if (needQuotes && quote != null) {
+                sb.append(quote);
+            }
+            first = false;
+        }
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/misc/comparator/Tuple2Comparator.java

@@ -0,0 +1,41 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.misc.comparator;
+
+import lombok.AllArgsConstructor;
+import org.nd4j.common.primitives.Pair;
+
+import java.io.Serializable;
+import java.util.Comparator;
+
+@AllArgsConstructor
+public class Tuple2Comparator<T> implements Comparator<Pair<T, Long>>, Serializable {
+
+    private final boolean ascending;
+
+    @Override
+    public int compare(Pair<T, Long> o1, Pair<T, Long> o2) {
+        if (ascending)
+            return Long.compare(o1.getSecond(), o2.getSecond());
+        else
+            return -Long.compare(o1.getSecond(), o2.getSecond());
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/rank/UnzipForCalculateSortedRankFunction.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.rank;
+
+import org.datavec.api.writable.LongWritable;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class UnzipForCalculateSortedRankFunction
+                implements Function<Pair<Pair<Writable, List<Writable>>, Long>, List<Writable>> {
+    @Override
+    public List<Writable> apply(Pair<Pair<Writable, List<Writable>>, Long> v1) {
+        List<Writable> inputWritables = new ArrayList<>(v1.getFirst().getSecond());
+        inputWritables.add(new LongWritable(v1.getSecond()));
+        return inputWritables;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/reduce/MapToPairForReducerFunction.java

@@ -0,0 +1,64 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.reduce;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.reduce.IAssociativeReducer;
+import org.datavec.api.transform.schema.Schema;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.List;
+
+
+@AllArgsConstructor
+public class MapToPairForReducerFunction implements Function<List<Writable>, Pair<String, List<Writable>>> {
+    public static final String GLOBAL_KEY = "";
+
+    private final IAssociativeReducer reducer;
+
+    @Override
+    public Pair<String, List<Writable>> apply(List<Writable> writables) {
+        List<String> keyColumns = reducer.getKeyColumns();
+
+        if(keyColumns == null){
+            //Global reduction
+            return Pair.of(GLOBAL_KEY, writables);
+        } else {
+            Schema schema = reducer.getInputSchema();
+            String key;
+            if (keyColumns.size() == 1)
+                key = writables.get(schema.getIndexOfColumn(keyColumns.get(0))).toString();
+            else {
+                StringBuilder sb = new StringBuilder();
+                for (int i = 0; i < keyColumns.size(); i++) {
+                    if (i > 0)
+                        sb.append("_");
+                    sb.append(writables.get(schema.getIndexOfColumn(keyColumns.get(i))).toString());
+                }
+                key = sb.toString();
+            }
+
+            return Pair.of(key, writables);
+        }
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/reduce/ReducerFunction.java

@@ -0,0 +1,44 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.reduce;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.reduce.IAssociativeReducer;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class ReducerFunction implements Function<Iterable<List<Writable>>, List<Writable>> {
+
+    private final IAssociativeReducer reducer;
+
+    @Override
+    public List<Writable> apply(Iterable<List<Writable>> lists) {
+
+        for (List<Writable> c : lists) {
+            reducer.aggregableReducer().accept(c);
+        }
+
+        return reducer.aggregableReducer().get();
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/ConvertToSequenceLengthOne.java

@@ -0,0 +1,34 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ConvertToSequenceLengthOne implements Function<List<Writable>, List<List<Writable>>> {
+    @Override
+    public List<List<Writable>> apply(List<Writable> writables) {
+        return Collections.singletonList(writables);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/LocalGroupToSequenceFunction.java

@@ -0,0 +1,47 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.sequence.SequenceComparator;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalGroupToSequenceFunction implements Function<List<List<Writable>>, List<List<Writable>>> {
+
+    private final SequenceComparator comparator;
+
+    @Override
+    public List<List<Writable>> apply(List<List<Writable>> lists) {
+
+        List<List<Writable>> list = new ArrayList<>();
+        list.addAll(lists);
+
+        Collections.sort(list, comparator);
+
+        return list;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/LocalMapToPairByColumnFunction.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalMapToPairByColumnFunction implements Function<List<Writable>, Pair<Writable, List<Writable>>> {
+
+    private final int keyColumnIdx;
+
+    @Override
+    public Pair<Writable, List<Writable>> apply(List<Writable> writables) {
+        return Pair.of(writables.get(keyColumnIdx), writables);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/LocalMapToPairByMultipleColumnsFunction.java

@@ -0,0 +1,45 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+import org.nd4j.common.primitives.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalMapToPairByMultipleColumnsFunction
+                implements Function<List<Writable>, Pair<List<Writable>, List<Writable>>> {
+
+    private final int[] keyColumnIdxs;
+
+    @Override
+    public Pair<List<Writable>, List<Writable>> apply(List<Writable> writables) {
+        List<Writable> keyOut = new ArrayList<>(keyColumnIdxs.length);
+        for (int keyColumnIdx : keyColumnIdxs) {
+            keyOut.add(writables.get(keyColumnIdx));
+        }
+        return Pair.of(keyOut, writables);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/LocalSequenceFilterFunction.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.filter.Filter;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalSequenceFilterFunction implements Function<List<List<Writable>>, Boolean> {
+
+    private final Filter filter;
+
+    @Override
+    public Boolean apply(List<List<Writable>> v1) {
+        return !filter.removeSequence(v1); //return true to keep example (Filter: return true to remove)
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/sequence/LocalSequenceTransformFunction.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.sequence;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.Transform;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalSequenceTransformFunction implements Function<List<List<Writable>>, List<List<Writable>>> {
+
+    private final Transform transform;
+
+    @Override
+    public List<List<Writable>> apply(List<List<Writable>> v1) {
+        return transform.mapSequence(v1);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/LocalTransformFunction.java

@@ -0,0 +1,51 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.datavec.api.transform.Transform;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.LocalTransformExecutor;
+import org.nd4j.common.function.Function;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@AllArgsConstructor
+@Slf4j
+public class LocalTransformFunction implements Function<List<Writable>, List<Writable>> {
+
+    private final Transform transform;
+
+    @Override
+    public List<Writable> apply(List<Writable> v1) {
+        if (LocalTransformExecutor.isTryCatch()) {
+            try {
+                return transform.map(v1);
+            } catch (Exception e) {
+                log.warn("Error occurred " + e + " on record " + v1);
+                return new ArrayList<>();
+            }
+        }
+        return transform.map(v1);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/LocalTransformProcessFunction.java

@@ -0,0 +1,35 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform;
+
+import org.datavec.api.transform.TransformProcess;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.BaseFlatMapFunctionAdaptee;
+
+import java.util.List;
+
+public class LocalTransformProcessFunction extends BaseFlatMapFunctionAdaptee<List<Writable>, List<Writable>> {
+
+    public LocalTransformProcessFunction(TransformProcess transformProcess) {
+        super(new LocalTransformProcessFunctionAdapter(transformProcess));
+    }
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/LocalTransformProcessFunctionAdapter.java

@@ -0,0 +1,46 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform;
+
+import org.datavec.api.transform.TransformProcess;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.functions.FlatMapFunctionAdapter;
+
+import java.util.Collections;
+import java.util.List;
+
+public class LocalTransformProcessFunctionAdapter implements FlatMapFunctionAdapter<List<Writable>, List<Writable>> {
+
+    private final TransformProcess transformProcess;
+
+    public LocalTransformProcessFunctionAdapter(TransformProcess transformProcess) {
+        this.transformProcess = transformProcess;
+    }
+
+    @Override
+    public List<List<Writable>> call(List<Writable> v1) throws Exception {
+        List<Writable> newList = transformProcess.execute(v1);
+        if (newList == null)
+            return Collections.emptyList(); //Example was filtered out
+        else
+            return Collections.singletonList(newList);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/SequenceSplitFunction.java

@@ -0,0 +1,35 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform;
+
+import org.datavec.api.transform.sequence.SequenceSplit;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.BaseFlatMapFunctionAdaptee;
+
+import java.util.List;
+
+public class SequenceSplitFunction extends BaseFlatMapFunctionAdaptee<List<List<Writable>>, List<List<Writable>>> {
+
+    public SequenceSplitFunction(SequenceSplit split) {
+        super(new SequenceSplitFunctionAdapter(split));
+    }
+
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/SequenceSplitFunctionAdapter.java

@@ -0,0 +1,42 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform;
+
+import org.datavec.api.transform.sequence.SequenceSplit;
+import org.datavec.api.writable.Writable;
+import org.datavec.local.transforms.functions.FlatMapFunctionAdapter;
+
+import java.util.List;
+
+public class SequenceSplitFunctionAdapter
+                implements FlatMapFunctionAdapter<List<List<Writable>>, List<List<Writable>>> {
+
+    private final SequenceSplit split;
+
+    public SequenceSplitFunctionAdapter(SequenceSplit split) {
+        this.split = split;
+    }
+
+    @Override
+    public List<List<List<Writable>>> call(List<List<Writable>> collections) throws Exception {
+        return split.split(collections);
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/filter/FilterWritablesBySchemaFunction.java

@@ -0,0 +1,63 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform.filter;
+
+import org.datavec.api.transform.metadata.ColumnMetaData;
+import org.datavec.api.writable.NullWritable;
+import org.datavec.api.writable.Text;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+public class FilterWritablesBySchemaFunction implements Function<Writable, Boolean> {
+
+    private final ColumnMetaData meta;
+    private final boolean keepValid; //If true: keep valid. If false: keep invalid
+    private final boolean excludeMissing; //If true: remove/exclude any
+
+
+    public FilterWritablesBySchemaFunction(ColumnMetaData meta, boolean keepValid) {
+        this(meta, keepValid, false);
+    }
+
+    /**
+     *
+     * @param meta              Column meta data
+     * @param keepValid         If true: keep only the valid writables. If false: keep only the invalid writables
+     * @param excludeMissing    If true: don't return any missing values, regardless of keepValid setting (i.e., exclude any NullWritable or empty string values)
+     */
+    public FilterWritablesBySchemaFunction(ColumnMetaData meta, boolean keepValid, boolean excludeMissing) {
+        this.meta = meta;
+        this.keepValid = keepValid;
+        this.excludeMissing = excludeMissing;
+    }
+
+    @Override
+    public Boolean apply(Writable v1) {
+        boolean valid = meta.isValid(v1);
+        if (excludeMissing && (v1 instanceof NullWritable
+                        || v1 instanceof Text && (v1.toString() == null || v1.toString().isEmpty())))
+            return false; //Remove
+        if (keepValid)
+            return valid; //return true to keep
+        else
+            return !valid;
+    }
+}

datavec/datavec-local/src/main/java/org/datavec/local/transforms/transform/filter/LocalFilterFunction.java

@@ -0,0 +1,39 @@
+/*
+ *  ******************************************************************************
+ *  *
+ *  *
+ *  * This program and the accompanying materials are made available under the
+ *  * terms of the Apache License, Version 2.0 which is available at
+ *  * https://www.apache.org/licenses/LICENSE-2.0.
+ *  *
+ *  *  See the NOTICE file distributed with this work for additional
+ *  *  information regarding copyright ownership.
+ *  * 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.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *  *****************************************************************************
+ */
+
+package org.datavec.local.transforms.transform.filter;
+
+import lombok.AllArgsConstructor;
+import org.datavec.api.transform.filter.Filter;
+import org.datavec.api.writable.Writable;
+import org.nd4j.common.function.Function;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class LocalFilterFunction implements Function<List<Writable>, Boolean> {
+
+    private final Filter filter;
+
+    @Override
+    public Boolean apply(List<Writable> v1) {
+        return !filter.removeExample(v1); //return true to keep example (Filter: return true to remove)
+    }
+}

datavec/datavec-local/src/main/java9/module-info.java

@@ -0,0 +1,24 @@
+open module datavec.local {
+    requires arrow.memory.core;
+    requires commons.io;
+    requires datavec.arrow;
+    requires guava;
+    requires protonpack;
+    requires slf4j.api;
+    requires datavec.api;
+    requires nd4j.api;
+    requires nd4j.common;
+    exports org.datavec.local.transforms;
+    exports org.datavec.local.transforms.analysis.aggregate;
+    exports org.datavec.local.transforms.analysis.histogram;
+    exports org.datavec.local.transforms.functions;
+    exports org.datavec.local.transforms.functions.data;
+    exports org.datavec.local.transforms.join;
+    exports org.datavec.local.transforms.misc;
+    exports org.datavec.local.transforms.misc.comparator;
+    exports org.datavec.local.transforms.rank;
+    exports org.datavec.local.transforms.reduce;
+    exports org.datavec.local.transforms.sequence;
+    exports org.datavec.local.transforms.transform;
+    exports org.datavec.local.transforms.transform.filter;
+}