fields =
definitions.stream()
.map(def -> Field.of(def.getName(), toBeamRowType(def.getType())).withNullable(true))
.map(
fld ->
keyOrder.containsKey(fld.getName())
? fld.withType(
fld.getType()
.withMetadata(KEY_ORDER_METADATA_KEY, keyOrder.get(fld.getName()).toString()))
: fld)
.collect(Collectors.toList());
return Schema.builder().addFields(fields).build();
}
/**
* This method takes Objects retrieved from Cassandra as well as their Column type and converts
* them into Objects that are supported by the Beam {@link Schema} and {@link Row}.
*
* The method starts with the complex types {@link List}, {@link Map} and {@link Set}. For all
* other types it assumes that they are primitives and passes them on.
*
* @param object The Cassandra cell value to be converted.
* @param type The Cassandra Data Type
* @return the beam compatible object.
*/
private Object mapValue(Object object, DataType type) {
DataType.Name typeName = type.getName();
if (typeName == Name.LIST) {
DataType innerType = type.getTypeArguments().get(0);
List list = (List) object;
// Apply toBeamObject on all items.
return list.stream()
.map(value -> toBeamObject(value, innerType))
.collect(Collectors.toList());
} else if (typeName == Name.MAP) {
DataType ktype = type.getTypeArguments().get(0);
DataType vtype = type.getTypeArguments().get(1);
Set map = ((Map) object).entrySet();
// Apply toBeamObject on both key and value.
return map.stream()
.collect(
Collectors.toMap(
e -> toBeamObject(e.getKey(), ktype), e -> toBeamObject(e.getValue(), vtype)));
} else if (typeName == Name.SET) {
DataType innerType = type.getTypeArguments().get(0);
List list = new ArrayList((Set) object);
// Apply toBeamObject on all items.
return list.stream().map(l -> toBeamObject(l, innerType)).collect(Collectors.toList());
} else {
return toBeamObject(object, type);
}
}
/**
* This method converts Cassandra {@link DataType} to Beam {@link FieldType}. Tuples are as of yet
* not supported as there is no corresponding type in the Beam {@link Schema}.
*
* @param type the Cassandra DataType to be converted
* @return the corresponding Beam Schema field type.
* @see org.apache.beam.sdk.schemas.Schema.FieldType
* @see com.datastax.driver.core.DataType
* @link Cassandra
* Tuple
*/
private FieldType toBeamRowType(DataType type) {
DataType.Name n = type.getName();
switch (n) {
case TIMESTAMP:
case DATE:
return FieldType.DATETIME;
case BLOB:
return FieldType.BYTES;
case BOOLEAN:
return FieldType.BOOLEAN;
case DECIMAL:
return FieldType.DECIMAL;
case DOUBLE:
return FieldType.DOUBLE;
case FLOAT:
return FieldType.FLOAT;
case INT:
return FieldType.INT32;
case VARINT:
return FieldType.DECIMAL;
case SMALLINT:
return FieldType.INT16;
case TINYINT:
return FieldType.BYTE;
case LIST:
case SET:
DataType innerType = type.getTypeArguments().get(0);
return FieldType.array(toBeamRowType(innerType));
case MAP:
DataType kDataType = type.getTypeArguments().get(0);
DataType vDataType = type.getTypeArguments().get(1);
FieldType k = toBeamRowType(kDataType);
FieldType v = toBeamRowType(vDataType);
return FieldType.map(k, v);
case VARCHAR:
case TEXT:
case INET:
case UUID:
case TIMEUUID:
case ASCII:
return FieldType.STRING;
case BIGINT:
case COUNTER:
case TIME:
return FieldType.INT64;
default:
throw new UnsupportedOperationException("Datatype " + type.getName() + " not supported.");
}
}
/**
* Most primitivies are represented the same way in Beam and Cassandra however there are a few
* that differ. This method converts the native representation of timestamps, uuids, varint, dates
* and IPs to a format which works for the Beam Schema.
*
* Dates and Timestamps are returned as DateTime objects whilst UUIDs are converted to Strings.
* Varint is converted into BigDecimal. The rest simply pass through as they are.
*
* @param value The object value as retrieved from Cassandra.
* @param typeName The Cassandra schema type.
* @see org.apache.beam.sdk.schemas.Schema.FieldType
* @return The corresponding representation that works in the Beam schema.
*/
private Object toBeamObject(Object value, DataType typeName) {
if (typeName == null || typeName.getName() == null) {
throw new UnsupportedOperationException(
"Unspecified Cassandra data type, cannot convert to beam row primitive.");
}
switch (typeName.getName()) {
case TIMESTAMP:
return new DateTime(value);
case UUID:
return ((UUID) value).toString();
case VARINT:
return new BigDecimal((BigInteger) value);
case TIMEUUID:
return ((UUID) value).toString();
case DATE:
LocalDate ld = (LocalDate) value;
return new DateTime(ld.getYear(), ld.getMonth(), ld.getDay(), 0, 0);
case INET:
return ((InetAddress) value).getHostAddress();
default:
return value;
}
}
}