Cast extends PTransform<PCollection<T>, PCollection<Row>> { @SuppressWarnings("unchecked") public static Object castValue(Object inputValue, FieldType input, FieldType output) { TypeName inputType = input.getTypeName(); TypeName outputType = output.getTypeName(); if (inputValue == null) { return null; } switch (inputType) { case ROW: return castRow((Row) inputValue, input.getRowSchema(), output.getRowSchema()); case ARRAY: case ITERABLE:; Iterable<Object> inputValues = (Iterable<Object>) inputValue; List<Object> outputValues = new ArrayList<>(Iterables.size(inputValues)); for (Object elem : inputValues) { outputValues.add( castValue(elem, input.getCollectionElementType(), output.getCollectionElementType())); } return outputValues; case MAP: Map<Object, Object> inputMap = (Map<Object, Object>) inputValue; Map<Object, Object> outputMap = Maps.newHashMapWithExpectedSize(inputMap.size()); for (Map.Entry<Object, Object> entry : inputMap.entrySet()) { Object outputKey = castValue(entry.getKey(), input.getMapKeyType(), output.getMapKeyType()); Object outputValue = castValue(entry.getValue(), input.getMapValueType(), output.getMapValueType()); outputMap.put(outputKey, outputValue); } return outputMap; default: if (inputType.equals(outputType)) { return inputValue; } if (inputType.isNumericType()) { return castNumber((Number) inputValue, inputType, outputType); } else { throw new IllegalArgumentException("input should be array, map, numeric or row"); } } }  abstract Schema outputSchema(); abstract Validator validator(); static Cast<T> of(Schema outputSchema, Validator validator); static Cast<T> widening(Schema outputSchema); static Cast<T> narrowing(Schema outputSchema); static boolean isIntegral(TypeName type); static boolean isDecimal(TypeName type); void verifyCompatibility(Schema inputSchema); @Override PCollection<Row> expand(PCollection<T> input); static Row castRow(Row input, Schema inputSchema, Schema outputSchema); static Number castNumber(Number value, TypeName input, TypeName output); @SuppressWarnings("unchecked") static Object castValue(Object inputValue, FieldType input, FieldType output);  }
@Test public void testCastArray() { Object output = Cast.castValue( Arrays.asList((short) 1, (short) 2, (short) 3), Schema.FieldType.array(Schema.FieldType.INT16), Schema.FieldType.array(Schema.FieldType.INT32)); assertEquals(Arrays.asList(1, 2, 3), output); }