VectorInteractionMapper extends MISOMapper { @Override protected Object map(Object[] input) { if (input.length != 2) { throw new IllegalArgumentException("VectorInteraction only support two input columns."); } if (input[0] == null || input[1] == null) { return null; } Vector vector1 = VectorUtil.getVector(input[0]); Vector vector2 = VectorUtil.getVector(input[1]); if (vector1 instanceof SparseVector) { if (vector2 instanceof DenseVector) { throw new IllegalArgumentException("Make sure the two input vectors are both dense or sparse."); } SparseVector sparseVector = (SparseVector) vector1; int vecSize = sparseVector.size(); int[] indices = sparseVector.getIndices(); double[] values = sparseVector.getValues(); SparseVector scalingVector = (SparseVector) vector2; int scalingSize = scalingVector.size(); int[] scalingIndices = scalingVector.getIndices(); double[] scalingValues = scalingVector.getValues(); double[] interactionValues = new double[scalingIndices.length * indices.length]; int[] interactionIndices = new int[scalingIndices.length * indices.length]; for (int i = 0; i < indices.length; ++i) { int idxBase = i * scalingIndices.length; for (int j = 0; j < scalingIndices.length; ++j) { int idx = idxBase + j; interactionIndices[idx] = vecSize * scalingIndices[j] + indices[i]; interactionValues[idx] = values[i] * scalingValues[j]; } } return new SparseVector(vecSize * scalingSize, interactionIndices, interactionValues); } else { if (vector2 instanceof SparseVector) { throw new IllegalArgumentException("Make sure the two input vectors are both dense or sparse."); } double[] vecArray = ((DenseVector) vector1).getData(); double[] scalingArray = ((DenseVector) vector2).getData(); DenseVector inter = new DenseVector(vecArray.length * scalingArray.length); double[] interArray = inter.getData(); for (int i = 0; i < vecArray.length; ++i) { int idxBase = i * scalingArray.length; for (int j = 0; j < scalingArray.length; ++j) { interArray[idxBase + j] = vecArray[i] * scalingArray[j]; } } return inter; } } VectorInteractionMapper(TableSchema dataSchema, Params params);   }
@Test public void testSparse() throws Exception { TableSchema schema = new TableSchema(new String[] {"c0", "c1"}, new TypeInformation <?>[] {Types.STRING, Types.STRING}); TableSchema outSchema = new TableSchema(new String[] {"c0", "out"}, new TypeInformation <?>[] {Types.STRING, VectorTypes.VECTOR}); Params params = new Params() .set(VectorInteractionParams.SELECTED_COLS, new String[] {"c0", "c1"}) .set(VectorInteractionParams.OUTPUT_COL, "out") .set(VectorInteractionParams.RESERVED_COLS, new String[] {"c0"}); VectorInteractionMapper mapper = new VectorInteractionMapper(schema, params); assertEquals(mapper.map(Row.of(new SparseVector(10, new int[]{0, 9}, new double[]{1.0, 4.0}), new SparseVector(10, new int[]{0, 9}, new double[]{1.0, 4.0}))).getField(1), new SparseVector(100, new int[]{0, 9, 90, 99}, new double[]{1.0, 4.0, 4.0, 16.0})); assertEquals(mapper.getOutputSchema(), outSchema); }
@Test public void testReserveTwoCol() throws Exception { TableSchema schema = new TableSchema(new String[] {"c0", "c1"}, new TypeInformation <?>[] {Types.STRING, Types.STRING}); TableSchema outSchema = new TableSchema(new String[] {"c0", "c1", "out"}, new TypeInformation <?>[] {Types.STRING, Types.STRING, VectorTypes.VECTOR}); Params params = new Params() .set(VectorInteractionParams.SELECTED_COLS, new String[] {"c0", "c1"}) .set(VectorInteractionParams.OUTPUT_COL, "out"); VectorInteractionMapper mapper = new VectorInteractionMapper(schema, params); assertEquals(mapper.map(Row.of(new DenseVector(new double[]{3.0, 4.0}), new DenseVector(new double[]{3.0, 4.0}))) .getField(2), new DenseVector(new double[]{9.0, 12.0, 12.0, 16.0})); assertEquals(mapper.getOutputSchema(), outSchema); }
@Test public void testReserveOneCol() throws Exception { TableSchema schema = new TableSchema(new String[] {"c0", "c1"}, new TypeInformation <?>[] {Types.STRING, Types.STRING}); TableSchema outSchema = new TableSchema(new String[] {"c0", "out"}, new TypeInformation <?>[] {Types.STRING, VectorTypes.VECTOR}); Params params = new Params() .set(VectorInteractionParams.SELECTED_COLS, new String[] {"c0", "c1"}) .set(VectorInteractionParams.OUTPUT_COL, "out") .set(VectorInteractionParams.RESERVED_COLS, new String[] {"c0"}); VectorInteractionMapper mapper = new VectorInteractionMapper(schema, params); assertEquals(mapper.map(Row.of(new DenseVector(new double[]{3.0, 4.0}), new DenseVector(new double[]{3.0, 4.0}))) .getField(1), new DenseVector(new double[]{9.0, 12.0, 12.0, 16.0})); assertEquals(mapper.getOutputSchema(), outSchema); }