You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by ac...@apache.org on 2016/04/15 07:23:58 UTC
incubator-systemml git commit: [SYSTEMML-560] Converters between CSV
and frame binaray block
Repository: incubator-systemml
Updated Branches:
refs/heads/master 6aef70cfb -> 98bd78f1a
[SYSTEMML-560] Converters between CSV and frame binaray block
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/98bd78f1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/98bd78f1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/98bd78f1
Branch: refs/heads/master
Commit: 98bd78f1a96f15a5518636db9a439f847199d1a1
Parents: 6aef70c
Author: Arvind Surve <ac...@yahoo.com>
Authored: Thu Apr 14 22:17:17 2016 -0700
Committer: Arvind Surve <ac...@yahoo.com>
Committed: Thu Apr 14 22:17:17 2016 -0700
----------------------------------------------------------------------
.../spark/utils/FrameRDDConverterUtils.java | 314 +++++++++++++++++++
.../functions/frame/FrameConverterTest.java | 250 +++++++++++++++
2 files changed, 564 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/98bd78f1/src/main/java/org/apache/sysml/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/utils/FrameRDDConverterUtils.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
new file mode 100644
index 0000000..6f0912a
--- /dev/null
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
@@ -0,0 +1,314 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysml.runtime.instructions.spark.utils;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+import org.apache.spark.api.java.function.PairFunction;
+
+import scala.Tuple2;
+
+import org.apache.sysml.parser.Expression.ValueType;
+import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.instructions.spark.data.SerLongWritable;
+import org.apache.sysml.runtime.instructions.spark.data.SerText;
+import org.apache.sysml.runtime.io.IOUtilFunctions;
+import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
+import org.apache.sysml.runtime.matrix.data.CSVFileFormatProperties;
+import org.apache.sysml.runtime.matrix.data.FrameBlock;
+
+public class FrameRDDConverterUtils
+{
+ /**
+ *
+ * @param in
+ * @param mcIn
+ * @param props
+ * @param strict
+ * @return
+ */
+ public static JavaRDD<String> binaryBlockToCsv(JavaPairRDD<LongWritable,FrameBlock> in, MatrixCharacteristics mcIn, CSVFileFormatProperties props, boolean strict)
+ {
+ JavaPairRDD<LongWritable,FrameBlock> input = in;
+
+ //sort if required (on blocks/rows)
+ if( strict ) {
+ input = input.sortByKey(true);
+ }
+
+ //convert binary block to csv (from blocks/rows)
+ JavaRDD<String> out = input
+ .flatMap(new BinaryBlockToCSVFunction(props));
+
+ return out;
+ }
+
+ /**
+ *
+ * @param sc
+ * @param input
+ * @param mcOut
+ * @param hasHeader
+ * @param delim
+ * @param fill
+ * @param missingValue
+ * @return
+ * @throws DMLRuntimeException
+ */
+ public static JavaPairRDD<LongWritable, FrameBlock> csvToBinaryBlock(JavaSparkContext sc,
+ JavaPairRDD<LongWritable, Text> input, MatrixCharacteristics mcOut,
+ boolean hasHeader, String delim, boolean fill, double fillValue)
+ throws DMLRuntimeException
+ {
+ //determine unknown dimensions and sparsity if required
+ if( !mcOut.dimsKnown(true) ) {
+ JavaRDD<String> tmp = input.values()
+ .map(new TextToStringFunction());
+ long rlen = tmp.count() - (hasHeader ? 1 : 0);
+ long clen = tmp.first().split(delim).length;
+ mcOut.set(rlen, clen, mcOut.getRowsPerBlock(), mcOut.getColsPerBlock(), -1);
+ }
+
+ //prepare csv w/ row indexes (sorted by filenames)
+ JavaPairRDD<Text,Long> prepinput = input.values()
+ .zipWithIndex(); //zip row index
+
+ //convert csv rdd to binary block rdd (w/ partial blocks)
+ JavaPairRDD<LongWritable, FrameBlock> out =
+ prepinput.mapPartitionsToPair(
+ new CSVToBinaryBlockFunction(mcOut, hasHeader, delim, fill));
+
+ return out;
+ }
+
+ /**
+ * @param sc
+ * @param input
+ * @param mcOut
+ * @param hasHeader
+ * @param delim
+ * @param fill
+ * @param fillValue
+ * @return
+ * @throws DMLRuntimeException
+ */
+ public static JavaPairRDD<LongWritable, FrameBlock> csvToBinaryBlock(JavaSparkContext sc,
+ JavaRDD<String> input, MatrixCharacteristics mcOut,
+ boolean hasHeader, String delim, boolean fill, double fillValue)
+ throws DMLRuntimeException
+ {
+ //convert string rdd to serializable longwritable/text
+ JavaPairRDD<LongWritable, Text> prepinput =
+ input.mapToPair(new StringToSerTextFunction());
+
+ //convert to binary block
+ return csvToBinaryBlock(sc, prepinput, mcOut, hasHeader, delim, fill, fillValue);
+ }
+
+ /**
+ *
+ */
+ private static class StringToSerTextFunction implements PairFunction<String, LongWritable, Text>
+ {
+ private static final long serialVersionUID = 8683232211035837695L;
+
+ @Override
+ public Tuple2<LongWritable, Text> call(String arg0)
+ throws Exception
+ {
+ SerLongWritable slarg = new SerLongWritable(1L);
+ SerText starg = new SerText(arg0);
+ return new Tuple2<LongWritable,Text>(slarg, starg);
+ }
+ }
+
+ /////////////////////////////////
+ // CSV-SPECIFIC FUNCTIONS
+
+ /**
+ *
+ */
+ private static class TextToStringFunction implements Function<Text,String>
+ {
+ private static final long serialVersionUID = -2744814934501782747L;
+
+ @Override
+ public String call(Text v1)
+ throws Exception
+ {
+ return v1.toString();
+ }
+
+ }
+
+ /**
+ * This functions allows to map rdd partitions of csv rows into a set of partial binary blocks.
+ *
+ * NOTE: For this csv to binary block function, we need to hold all output blocks per partition
+ * in-memory. Hence, we keep state of all column blocks and aggregate row segments into these blocks.
+ * In terms of memory consumption this is better than creating partial blocks of row segments.
+ *
+ */
+ private static class CSVToBinaryBlockFunction implements PairFlatMapFunction<Iterator<Tuple2<Text,Long>>,LongWritable,FrameBlock>
+ {
+ private static final long serialVersionUID = -1976803898174960086L;
+
+ private long _clen = -1;
+ private boolean _hasHeader = false;
+ private String _delim = null;
+ private boolean _fill = false;
+ private int _maxRowsPerBlock = -1;
+
+ protected static final int BUFFER_SIZE = 2 * 1000 * 1000; //2M elements
+
+
+ public CSVToBinaryBlockFunction(MatrixCharacteristics mc, boolean hasHeader, String delim, boolean fill)
+ {
+ _clen = mc.getCols();
+ _hasHeader = hasHeader;
+ _delim = delim;
+ _fill = fill;
+ _maxRowsPerBlock = Math.max((int) (BUFFER_SIZE/_clen), 1);
+ }
+
+ @Override
+ public Iterable<Tuple2<LongWritable, FrameBlock>> call(Iterator<Tuple2<Text,Long>> arg0)
+ throws Exception
+ {
+ ArrayList<Tuple2<LongWritable,FrameBlock>> ret = new ArrayList<Tuple2<LongWritable,FrameBlock>>();
+
+ LongWritable[] ix = new LongWritable[1];
+ FrameBlock[] mb = new FrameBlock[1];
+ int iRowsInBlock = 0;
+
+ while( arg0.hasNext() )
+ {
+ Tuple2<Text,Long> tmp = arg0.next();
+ String row = tmp._1().toString();
+ long rowix = tmp._2();
+ if(!_hasHeader) // In case there is no header, rowindex to be adjusted to base 1.
+ ++rowix;
+ if(_hasHeader && rowix == 0) //Skip header
+ continue;
+
+ if( iRowsInBlock == 0 || iRowsInBlock == _maxRowsPerBlock) {
+ if( iRowsInBlock == _maxRowsPerBlock )
+ flushBlocksToList(ix, mb, ret);
+ createBlocks(rowix, ix, mb);
+ iRowsInBlock = 0;
+ }
+
+ //process row data
+ String[] parts = IOUtilFunctions.split(row, _delim);
+ boolean emptyFound = false;
+ mb[0].appendRow(parts);
+ ++iRowsInBlock;
+
+ //sanity check empty cells filled w/ values
+ IOUtilFunctions.checkAndRaiseErrorCSVEmptyField(row, _fill, emptyFound);
+ }
+
+ //flush last blocks
+ flushBlocksToList(ix, mb, ret);
+
+ return ret;
+ }
+
+ // Creates new state of empty column blocks for current global row index.
+ private void createBlocks(long rowix, LongWritable[] ix, FrameBlock[] mb)
+ {
+ //compute row block index and number of column blocks
+ ix[0] = new LongWritable(rowix);
+ mb[0] = new FrameBlock((int)_clen, ValueType.STRING);
+ }
+
+ // Flushes current state of filled column blocks to output list.
+ private void flushBlocksToList( LongWritable[] ix, FrameBlock[] mb, ArrayList<Tuple2<LongWritable,FrameBlock>> ret )
+ throws DMLRuntimeException
+ {
+ int len = ix.length;
+ for( int i=0; i<len; i++ )
+ if( mb[i] != null ) {
+ ret.add(new Tuple2<LongWritable,FrameBlock>(ix[i],mb[i]));
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private static class BinaryBlockToCSVFunction implements FlatMapFunction<Tuple2<LongWritable,FrameBlock>,String>
+ {
+ private static final long serialVersionUID = 8020608184930291069L;
+
+ private CSVFileFormatProperties _props = null;
+
+ public BinaryBlockToCSVFunction(CSVFileFormatProperties props) {
+ _props = props;
+ }
+
+ @Override
+ public Iterable<String> call(Tuple2<LongWritable, FrameBlock> arg0)
+ throws Exception
+ {
+ LongWritable ix = arg0._1();
+ FrameBlock blk = arg0._2();
+
+ ArrayList<String> ret = new ArrayList<String>();
+
+ //handle header information
+ if(_props.hasHeader() && ix.get()==1 ) {
+ StringBuilder sb = new StringBuilder();
+ for(int j = 1; j <= blk.getNumColumns(); j++) {
+ if(j != 1)
+ sb.append(_props.getDelim());
+ sb.append("C" + j);
+ }
+ ret.add(sb.toString());
+ }
+
+ //handle Frame block data
+ StringBuilder sb = new StringBuilder();
+ for(int i=0; i<blk.getNumRows(); i++) {
+ for(int j=0; j<blk.getNumColumns(); j++) {
+ if(j != 0)
+ sb.append(_props.getDelim());
+ Object val = blk.get(i, j);
+
+ if(val != null)
+ sb.append(val);
+ }
+ ret.add(sb.toString());
+ sb.setLength(0); //reset
+ }
+
+ return ret;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/98bd78f1/src/test/java/org/apache/sysml/test/integration/functions/frame/FrameConverterTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/sysml/test/integration/functions/frame/FrameConverterTest.java b/src/test/java/org/apache/sysml/test/integration/functions/frame/FrameConverterTest.java
new file mode 100644
index 0000000..590885e
--- /dev/null
+++ b/src/test/java/org/apache/sysml/test/integration/functions/frame/FrameConverterTest.java
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysml.test.integration.functions.frame;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.sysml.api.DMLScript;
+import org.apache.sysml.api.DMLScript.RUNTIME_PLATFORM;
+import org.apache.sysml.conf.ConfigurationManager;
+import org.apache.sysml.parser.Expression.ValueType;
+import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.instructions.cp.AppendCPInstruction.AppendType;
+import org.apache.sysml.runtime.instructions.spark.utils.FrameRDDConverterUtils;
+import org.apache.sysml.runtime.io.FrameReader;
+import org.apache.sysml.runtime.io.FrameReaderFactory;
+import org.apache.sysml.runtime.io.FrameWriter;
+import org.apache.sysml.runtime.io.FrameWriterFactory;
+import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
+import org.apache.sysml.runtime.matrix.data.CSVFileFormatProperties;
+import org.apache.sysml.runtime.matrix.data.FrameBlock;
+import org.apache.sysml.runtime.matrix.data.OutputInfo;
+import org.apache.sysml.runtime.util.MapReduceTool;
+import org.apache.sysml.runtime.util.UtilFunctions;
+import org.apache.sysml.test.integration.AutomatedTestBase;
+import org.apache.sysml.test.utils.TestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class FrameConverterTest extends AutomatedTestBase
+{
+ public enum ExecType { CP, CP_FILE, MR, SPARK, INVALID };
+
+ private final static String TEST_DIR = "functions/frame/io/";
+
+ private final static int rows = 1593;
+ private final static ValueType[] schemaStrings = new ValueType[]{ValueType.STRING, ValueType.STRING, ValueType.STRING};
+ private final static ValueType[] schemaMixed = new ValueType[]{ValueType.STRING, ValueType.DOUBLE, ValueType.INT, ValueType.BOOLEAN};
+
+ private final static String DELIMITER = "::";
+ private final static boolean HEADER = true;
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ }
+
+ @Test
+ public void testFrameStringsStringsCBindSp() {
+ runFrameCopyTest(schemaStrings, schemaStrings, AppendType.CBIND);
+ }
+
+ @Test
+ public void testFrameStringsStringsRBindSp() { //note: ncol(A)=ncol(B)
+ runFrameCopyTest(schemaStrings, schemaStrings, AppendType.RBIND);
+ }
+
+ @Test
+ public void testFrameMixedStringsCBindSp() {
+ runFrameCopyTest(schemaMixed, schemaStrings, AppendType.CBIND);
+ }
+
+ @Test
+ public void testFrameStringsMixedCBindSp() {
+ runFrameCopyTest(schemaStrings, schemaMixed, AppendType.CBIND);
+ }
+
+ @Test
+ public void testFrameMixedMixedCBindSp() {
+ runFrameCopyTest(schemaMixed, schemaMixed, AppendType.CBIND);
+ }
+
+ @Test
+ public void testFrameMixedMixedRBindSp() { //note: ncol(A)=ncol(B)
+ runFrameCopyTest(schemaMixed, schemaMixed, AppendType.RBIND);
+ }
+
+
+ /**
+ *
+ * @param sparseM1
+ * @param sparseM2
+ * @param instType
+ */
+ private void runFrameCopyTest( ValueType[] schema1, ValueType[] schema2, AppendType atype)
+ {
+ RUNTIME_PLATFORM platformOld = rtplatform;
+ rtplatform = RUNTIME_PLATFORM.SPARK;
+ boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
+ if( rtplatform == RUNTIME_PLATFORM.SPARK )
+ DMLScript.USE_LOCAL_SPARK_CONFIG = true;
+
+ try
+ {
+ //data generation
+ double[][] A = getRandomMatrix(rows, schema1.length, -10, 10, 0.9, 2373);
+ double[][] B = getRandomMatrix(rows, schema2.length, -10, 10, 0.9, 129);
+
+ //Initialize the frame data.
+ //init data frame 1
+ List<ValueType> lschema1 = Arrays.asList(schema1);
+ FrameBlock frame1 = new FrameBlock(lschema1);
+ initFrameData(frame1, A, lschema1);
+
+ //init data frame 2
+ List<ValueType> lschema2 = Arrays.asList(schema2);
+ FrameBlock frame2 = new FrameBlock(lschema2);
+ initFrameData(frame2, B, lschema2);
+
+ //Write frame data to disk
+ CSVFileFormatProperties fprop = new CSVFileFormatProperties();
+ fprop.setDelim(DELIMITER);
+ fprop.setHeader(HEADER);
+
+ //Verify CSV data through distributed environment (Spark)
+
+ ////Verify CSV data format
+ writeAndVerifyDistData(OutputInfo.CSVOutputInfo, frame1, fprop);
+ writeAndVerifyDistData(OutputInfo.CSVOutputInfo, frame2, fprop);
+ }
+ catch(Exception ex) {
+ ex.printStackTrace();
+ throw new RuntimeException(ex);
+ }
+ finally
+ {
+ rtplatform = platformOld;
+ DMLScript.USE_LOCAL_SPARK_CONFIG = sparkConfigOld;
+ }
+ }
+
+ void initFrameData(FrameBlock frame, double[][] data, List<ValueType> lschema)
+ {
+ Object[] row1 = new Object[lschema.size()];
+ for( int i=0; i<rows; i++ ) {
+ for( int j=0; j<lschema.size(); j++ )
+ data[i][j] = UtilFunctions.objectToDouble(lschema.get(j),
+ row1[j] = UtilFunctions.doubleToObject(lschema.get(j), data[i][j]));
+ frame.appendRow(row1);
+ }
+ }
+
+ void verifyFrameData(FrameBlock frame1, FrameBlock frame2)
+ {
+ List<ValueType> lschema = frame1.getSchema();
+ for ( int i=0; i<frame1.getNumRows(); ++i )
+ for( int j=0; j<lschema.size(); j++ ) {
+ if( UtilFunctions.compareTo(lschema.get(j), frame1.get(i, j), frame2.get(i, j)) != 0)
+ Assert.fail("Target value for cell ("+ i + "," + j + ") is " + frame1.get(i, j) +
+ ", is not same as original value " + frame2.get(i, j));
+ }
+ }
+
+
+
+ /**
+ * @param oinfo
+ * @param frame1
+ * @param frame2
+ * @param fprop
+ * @return
+ * @throws DMLRuntimeException, IOException
+ */
+
+ void writeAndVerifyDistData(OutputInfo oinfo, FrameBlock frame, CSVFileFormatProperties fprop)
+ throws DMLRuntimeException, IOException
+ {
+ String fname = TEST_DIR + "/frameData";
+ String fnameVerify = TEST_DIR + "/frameDataVerify";
+
+ //Create writer
+ FrameWriter writer = FrameWriterFactory.createFrameWriter(oinfo, fprop);
+
+ //Write frame data to disk
+ writer.writeFrameToHDFS(frame, fname, frame.getNumRows(), frame.getNumColumns());
+
+ SparkConf conf = new SparkConf().setAppName("Frame").setMaster("local");
+ conf.set("spark.kryo.classesToRegister", "org.apache.hadoop.io.LongWritable");
+
+ try
+ {
+ conf.registerKryoClasses(new Class<?>[]{
+ Class.forName("org.apache.hadoop.io.LongWritable")
+ });
+ } catch (Exception e)
+ {
+ System.out.println("Register class exception: " + e);
+ }
+
+ JavaSparkContext sc = new JavaSparkContext(conf);
+
+ boolean hasHeader = HEADER;
+ String delim = "::";
+ boolean fill = false; //TODO: Do we need fill/fillValue?
+ double fillValue = 0.0;
+ MatrixCharacteristics mc = new MatrixCharacteristics();
+ mc.set(frame.getNumRows(), frame.getNumColumns(), ConfigurationManager.getBlocksize(), frame.getNumColumns());
+
+ JavaRDD<String> rddStrCsv1 = sc.textFile(fname);
+
+ JavaPairRDD<LongWritable, FrameBlock> pairRDDCsv1 =
+ FrameRDDConverterUtils.csvToBinaryBlock(sc, rddStrCsv1, mc, hasHeader, delim, fill, fillValue);
+ writer.writeFrameToHDFS(frame, fname, frame.getNumRows(), frame.getNumColumns());
+
+ JavaRDD<String> rddStrCsv1Verify = FrameRDDConverterUtils.binaryBlockToCsv(pairRDDCsv1, mc, fprop, true);
+
+ MapReduceTool.deleteFileIfExistOnHDFS(fnameVerify);
+ rddStrCsv1Verify.saveAsTextFile(fnameVerify);
+
+ //Create reader
+ FrameReader reader = FrameReaderFactory.createFrameReader(OutputInfo.getMatchingInputInfo(oinfo), fprop);
+
+ //Read frame data from disk
+ FrameBlock frameRead = reader.readFrameFromHDFS(fname, frame.getSchema(), frame.getNumRows(), frame.getNumColumns());
+ FrameBlock frameVerifyRead = reader.readFrameFromHDFS(fnameVerify, frame.getSchema(), frame.getNumRows(), frame.getNumColumns());
+
+ // Verify that data read with original frames
+ verifyFrameData(frame, frameRead);
+ verifyFrameData(frame, frameVerifyRead);
+
+ // Do cleanup
+ MapReduceTool.deleteFileIfExistOnHDFS(fname);
+ MapReduceTool.deleteFileIfExistOnHDFS(fnameVerify);
+
+ sc.close();
+ }
+}