com.datastax.driver.core.DataType#getName ( )源码实例Demo

/**
 * 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.
 *
 * <p>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;
  }
}
 

public static Object parseValue(final DataType type, final String value) {
  if (value == null) {
    return null;
  }
  switch (type.getName()) {
  case TEXT:
  case VARCHAR:
  case ASCII:
    return value;
  case INET:
    return type.parse("'" + value + "'");
  case INT:
  case VARINT:
  case BIGINT:
  case FLOAT:
  case DOUBLE:
  case DECIMAL:
  case BOOLEAN:
    return type.parse(value.replaceAll("\\s+", ""));
  default:
    return type.parse(value);
  }
}
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();
        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }
 

private static int getNumBytes(BoundStatement boundStatement, int i, DataType dataType) {
    switch (dataType.getName()) {
        case VARCHAR:
            String s = boundStatement.getString(i);
            return s == null ? 0 : s.length();
        case BLOB:
            ByteBuffer bb = boundStatement.getBytes(i);
            return bb == null ? 0 : bb.limit();
        default:
            return 0;
    }
}
 

DataType.Name checkType(int i, DataType.Name name1, DataType.Name name2) {
    DataType defined = getType(i);
    if (name1 != defined.getName() && name2 != defined.getName())
        throw new InvalidTypeException(String.format("Column %s is of type %s", getName(i), defined));

    return defined.getName();
}
 

DataType.Name checkType(int i, DataType.Name name1, DataType.Name name2, DataType.Name name3) {
    DataType defined = getType(i);
    if (name1 != defined.getName() && name2 != defined.getName() && name3 != defined.getName())
        throw new InvalidTypeException(String.format("Column %s is of type %s", getName(i), defined));

    return defined.getName();
}
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();
        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();
        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }
 

/**
 * This method wraps result set data in to DataEntry and creates a list of DataEntry.
 *
 * @param tableName         Table Name
 * @param row               Row
 * @param columnDefinitions Column Definition
 * @return DataEntry
 * @throws ODataServiceFault
 */
private ODataEntry createDataEntryFromRow(String tableName, Row row, ColumnDefinitions columnDefinitions)
        throws ODataServiceFault {
    String paramValue;
    ODataEntry entry = new ODataEntry();
    //Creating a unique string to represent the
    try {
        for (int i = 0; i < columnDefinitions.size(); i++) {
            String columnName = columnDefinitions.getName(i);
            DataType columnType = columnDefinitions.getType(i);

            switch (columnType.getName()) {
                case ASCII:
                    paramValue = row.getString(i);
                    break;
                case BIGINT:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getLong(i));
                    break;
                case BLOB:
                    paramValue = this.base64EncodeByteBuffer(row.getBytes(i));
                    break;
                case BOOLEAN:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getBool(i));
                    break;
                case COUNTER:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getLong(i));
                    break;
                case DECIMAL:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getDecimal(i));
                    break;
                case DOUBLE:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getDouble(i));
                    break;
                case FLOAT:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getFloat(i));
                    break;
                case INET:
                    paramValue = row.getInet(i).toString();
                    break;
                case INT:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getInt(i));
                    break;
                case TEXT:
                    paramValue = row.getString(i);
                    break;
                case TIMESTAMP:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getDate(i));
                    break;
                case UUID:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getUUID(i));
                    break;
                case VARCHAR:
                    paramValue = row.getString(i);
                    break;
                case VARINT:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getVarint(i));
                    break;
                case TIMEUUID:
                    paramValue = row.isNull(i) ? null : ConverterUtil.convertToString(row.getUUID(i));
                    break;
                case LIST:
                    paramValue = row.isNull(i) ? null : Arrays.toString(row.getList(i, Object.class).toArray());
                    break;
                case SET:
                    paramValue = row.isNull(i) ? null : row.getSet(i, Object.class).toString();
                    break;
                case MAP:
                    paramValue = row.isNull(i) ? null : row.getMap(i, Object.class, Object.class).toString();
                    break;
                case UDT:
                    paramValue = row.isNull(i) ? null : row.getUDTValue(i).toString();
                    break;
                case TUPLE:
                    paramValue = row.isNull(i) ? null : row.getTupleValue(i).toString();
                    break;
                case CUSTOM:
                    paramValue = row.isNull(i) ? null : this.base64EncodeByteBuffer(row.getBytes(i));
                    break;
                default:
                    paramValue = row.getString(i);
                    break;
            }
            entry.addValue(columnName, paramValue);
        }
    } catch (DataServiceFault e) {
        throw new ODataServiceFault(e, "Error occurred when creating OData entry. :" + e.getMessage());
    }
    //Set E-Tag to the entity
    entry.addValue("ETag", ODataUtils.generateETag(this.configID, tableName, entry));
    return entry;
}
 

/**
 * Creates an Avro schema from the given result set. The metadata (column definitions, data types, etc.) is used
 * to determine a schema for Avro.
 *
 * @param rs The result set from which an Avro schema will be created
 * @return An Avro schema corresponding to the given result set's metadata
 * @throws IOException If an error occurs during schema discovery/building
 */
public static Schema createSchema(final ResultSet rs) throws IOException {
    final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions();
    final int nrOfColumns = (columnDefinitions == null ? 0 : columnDefinitions.size());
    String tableName = "NiFi_Cassandra_Query_Record";
    if (nrOfColumns > 0) {
        String tableNameFromMeta = columnDefinitions.getTable(1);
        if (!StringUtils.isBlank(tableNameFromMeta)) {
            tableName = tableNameFromMeta;
        }
    }

    final SchemaBuilder.FieldAssembler<Schema> builder = SchemaBuilder.record(tableName).namespace("any.data").fields();
    if (columnDefinitions != null) {
        for (int i = 0; i < nrOfColumns; i++) {

            DataType dataType = columnDefinitions.getType(i);
            if (dataType == null) {
                throw new IllegalArgumentException("No data type for column[" + i + "] with name " + columnDefinitions.getName(i));
            }

            // Map types from Cassandra to Avro where possible
            if (dataType.isCollection()) {
                List<DataType> typeArguments = dataType.getTypeArguments();
                if (typeArguments == null || typeArguments.size() == 0) {
                    throw new IllegalArgumentException("Column[" + i + "] " + dataType.getName()
                            + " is a collection but no type arguments were specified!");
                }
                // Get the first type argument, to be used for lists and sets
                DataType firstArg = typeArguments.get(0);
                if (dataType.equals(DataType.set(firstArg))
                        || dataType.equals(DataType.list(firstArg))) {
                    builder.name(columnDefinitions.getName(i)).type().unionOf().nullBuilder().endNull().and().array()
                            .items(getUnionFieldType(getPrimitiveAvroTypeFromCassandraType(firstArg))).endUnion().noDefault();
                } else {
                    // Must be an n-arg collection like map
                    DataType secondArg = typeArguments.get(1);
                    if (dataType.equals(DataType.map(firstArg, secondArg))) {
                        builder.name(columnDefinitions.getName(i)).type().unionOf().nullBuilder().endNull().and().map().values(
                                getUnionFieldType(getPrimitiveAvroTypeFromCassandraType(secondArg))).endUnion().noDefault();
                    }
                }
            } else {
                builder.name(columnDefinitions.getName(i))
                        .type(getUnionFieldType(getPrimitiveAvroTypeFromCassandraType(dataType))).noDefault();
            }
        }
    }
    return builder.endRecord();
}
 

protected static Object getCassandraObject(Row row, int i, DataType dataType) {
    if (dataType.equals(DataType.blob())) {
        return row.getBytes(i);

    } else if (dataType.equals(DataType.varint()) || dataType.equals(DataType.decimal())) {
        // Avro can't handle BigDecimal and BigInteger as numbers - it will throw an
        // AvroRuntimeException such as: "Unknown datum type: java.math.BigDecimal: 38"
        return row.getObject(i).toString();

    } else if (dataType.equals(DataType.cboolean())) {
        return row.getBool(i);

    } else if (dataType.equals(DataType.cint())) {
        return row.getInt(i);

    } else if (dataType.equals(DataType.bigint())
            || dataType.equals(DataType.counter())) {
        return row.getLong(i);

    } else if (dataType.equals(DataType.ascii())
            || dataType.equals(DataType.text())
            || dataType.equals(DataType.varchar())) {
        return row.getString(i);

    } else if (dataType.equals(DataType.cfloat())) {
        return row.getFloat(i);

    } else if (dataType.equals(DataType.cdouble())) {
        return row.getDouble(i);

    } else if (dataType.equals(DataType.timestamp())) {
        return row.getTimestamp(i);

    } else if (dataType.equals(DataType.date())) {
        return row.getDate(i);

    } else if (dataType.equals(DataType.time())) {
        return row.getTime(i);

    } else if (dataType.isCollection()) {

        List<DataType> typeArguments = dataType.getTypeArguments();
        if (typeArguments == null || typeArguments.size() == 0) {
            throw new IllegalArgumentException("Column[" + i + "] " + dataType.getName()
                    + " is a collection but no type arguments were specified!");
        }
        // Get the first type argument, to be used for lists and sets (and the first in a map)
        DataType firstArg = typeArguments.get(0);
        TypeCodec firstCodec = codecRegistry.codecFor(firstArg);
        if (dataType.equals(DataType.set(firstArg))) {
            return row.getSet(i, firstCodec.getJavaType());
        } else if (dataType.equals(DataType.list(firstArg))) {
            return row.getList(i, firstCodec.getJavaType());
        } else {
            // Must be an n-arg collection like map
            DataType secondArg = typeArguments.get(1);
            TypeCodec secondCodec = codecRegistry.codecFor(secondArg);
            if (dataType.equals(DataType.map(firstArg, secondArg))) {
                return row.getMap(i, firstCodec.getJavaType(), secondCodec.getJavaType());
            }
        }

    } else {
        // The different types that we support are numbers (int, long, double, float),
        // as well as boolean values and Strings. Since Avro doesn't provide
        // timestamp types, we want to convert those to Strings. So we will cast anything other
        // than numbers or booleans to strings by using the toString() method.
        return row.getObject(i).toString();
    }
    return null;
}
 

@Override
public void activate(Context.OperatorContext context)
{
  com.datastax.driver.core.ResultSet rs
      = store.getSession().execute("select * from " + store.keyspace + "." + tablename);
  final ColumnDefinitions rsMetaData = rs.getColumnDefinitions();

  if (fieldInfos == null) {
    populateFieldInfosFromPojo(rsMetaData);
  }

  for (FieldInfo fieldInfo : getFieldInfos()) {
    // get the designated column's data type.
    final DataType type = rsMetaData.getType(fieldInfo.getColumnName());
    columnDataTypes.add(type);
    final Object getter;
    final String getterExpr = fieldInfo.getPojoFieldExpression();
    switch (type.getName()) {
      case ASCII:
      case TEXT:
      case VARCHAR:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, String.class);
        break;
      case BOOLEAN:
        getter = PojoUtils.createGetterBoolean(pojoClass, getterExpr);
        break;
      case INT:
        getter = PojoUtils.createGetterInt(pojoClass, getterExpr);
        break;
      case BIGINT:
      case COUNTER:
        getter = PojoUtils.createGetterLong(pojoClass, getterExpr);
        break;
      case FLOAT:
        getter = PojoUtils.createGetterFloat(pojoClass, getterExpr);
        break;
      case DOUBLE:
        getter = PojoUtils.createGetterDouble(pojoClass, getterExpr);
        break;
      case DECIMAL:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, BigDecimal.class);
        break;
      case SET:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, Set.class);
        break;
      case MAP:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, Map.class);
        break;
      case LIST:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, List.class);
        break;
      case TIMESTAMP:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, Date.class);
        break;
      case UUID:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, UUID.class);
        break;
      default:
        getter = PojoUtils.createGetter(pojoClass, getterExpr, Object.class);
        break;
    }
    getters.add(getter);
  }
  super.activate(context);
}
 

@Override
@SuppressWarnings("unchecked")
protected Statement setStatementParameters(PreparedStatement updateCommand, Object tuple) throws DriverException
{
  final BoundStatement boundStmnt = new BoundStatement(updateCommand);
  final int size = columnDataTypes.size();
  for (int i = 0; i < size; i++) {
    final DataType type = columnDataTypes.get(i);
    switch (type.getName()) {
      case UUID:
        final UUID id = ((Getter<Object, UUID>)getters.get(i)).get(tuple);
        boundStmnt.setUUID(i, id);
        break;
      case ASCII:
      case VARCHAR:
      case TEXT:
        final String ascii = ((Getter<Object, String>)getters.get(i)).get(tuple);
        boundStmnt.setString(i, ascii);
        break;
      case BOOLEAN:
        final boolean bool = ((GetterBoolean<Object>)getters.get(i)).get(tuple);
        boundStmnt.setBool(i, bool);
        break;
      case INT:
        final int intValue = ((GetterInt<Object>)getters.get(i)).get(tuple);
        boundStmnt.setInt(i, intValue);
        break;
      case BIGINT:
      case COUNTER:
        final long longValue = ((GetterLong<Object>)getters.get(i)).get(tuple);
        boundStmnt.setLong(i, longValue);
        break;
      case FLOAT:
        final float floatValue = ((GetterFloat<Object>)getters.get(i)).get(tuple);
        boundStmnt.setFloat(i, floatValue);
        break;
      case DOUBLE:
        final double doubleValue = ((GetterDouble<Object>)getters.get(i)).get(tuple);
        boundStmnt.setDouble(i, doubleValue);
        break;
      case DECIMAL:
        final BigDecimal decimal = ((Getter<Object, BigDecimal>)getters.get(i)).get(tuple);
        boundStmnt.setDecimal(i, decimal);
        break;
      case SET:
        Set<?> set = ((Getter<Object, Set<?>>)getters.get(i)).get(tuple);
        boundStmnt.setSet(i, set);
        break;
      case MAP:
        final Map<?,?> map = ((Getter<Object, Map<?,?>>)getters.get(i)).get(tuple);
        boundStmnt.setMap(i, map);
        break;
      case LIST:
        final List<?> list = ((Getter<Object, List<?>>)getters.get(i)).get(tuple);
        boundStmnt.setList(i, list);
        break;
      case TIMESTAMP:
        final Date date = ((Getter<Object, Date>)getters.get(i)).get(tuple);
        boundStmnt.setDate(i, LocalDate.fromMillisSinceEpoch(date.getTime()));
        break;
      default:
        throw new RuntimeException("unsupported data type " + type.getName());
    }
  }
  return boundStmnt;
}
 

@Override
public void activate(OperatorContext context)
{
  Long keyToken;
  if (startRow != null) {
    if ((keyToken = fetchKeyTokenFromDB(startRow)) != null) {
      startRowToken = keyToken;
    }
  }

  com.datastax.driver.core.ResultSet rs = store.getSession().execute("select * from " + store.keyspace + "." + tablename + " LIMIT " + 1);
  ColumnDefinitions rsMetaData = rs.getColumnDefinitions();

  if (query.contains("%t")) {
    query = query.replace("%t", tablename);
  }
  if (query.contains("%p")) {
    query = query.replace("%p", primaryKeyColumn);
  }
  if (query.contains("%l")) {
    query = query.replace("%l", limit + "");
  }

  LOG.debug("query is {}", query);

  for (FieldInfo fieldInfo : fieldInfos) {
    // Get the designated column's data type.
    DataType type = rsMetaData.getType(fieldInfo.getColumnName());
    columnDataTypes.add(type);
    Object setter;
    final String setterExpr = fieldInfo.getPojoFieldExpression();
    switch (type.getName()) {
      case ASCII:
      case TEXT:
      case VARCHAR:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, String.class);
        break;
      case BOOLEAN:
        setter = PojoUtils.createSetterBoolean(pojoClass, setterExpr);
        break;
      case INT:
        setter = PojoUtils.createSetterInt(pojoClass, setterExpr);
        break;
      case BIGINT:
      case COUNTER:
        setter = PojoUtils.createSetterLong(pojoClass, setterExpr);
        break;
      case FLOAT:
        setter = PojoUtils.createSetterFloat(pojoClass, setterExpr);
        break;
      case DOUBLE:
        setter = PojoUtils.createSetterDouble(pojoClass, setterExpr);
        break;
      case DECIMAL:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, BigDecimal.class);
        break;
      case SET:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, Set.class);
        break;
      case MAP:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, Map.class);
        break;
      case LIST:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, List.class);
        break;
      case TIMESTAMP:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, Date.class);
        break;
      case UUID:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, UUID.class);
        break;
      default:
        setter = PojoUtils.createSetter(pojoClass, setterExpr, Object.class);
        break;
    }
    setters.add(setter);
  }
}
 

/**
 * Resolves a getter that can be associated with the given field name in the Pojo matching to the given
 * data type of cassandra
 * @param tuplePayloadClass The tuple class that is used to build the getter from
 * @param getterExpr The name of the field representing the getter that needs to be generated
 * @param returnDataTypeOfGetter The Data type of the cassandra column
 * @param userDefinedTypesClass A map that can provide for a UDT class given a column name
 * @return The getter object that can be used to extract the value at runtime
 */
public static Object resolveGetterForField(Class tuplePayloadClass, String getterExpr,
    DataType returnDataTypeOfGetter, Map<String,Class> userDefinedTypesClass)
{
  Object getter = null;
  switch (returnDataTypeOfGetter.getName()) {
    case INT:
      getter = PojoUtils.createGetterInt(tuplePayloadClass, getterExpr);
      break;
    case BIGINT:
    case COUNTER:
      getter = PojoUtils.createGetterLong(tuplePayloadClass, getterExpr);
      break;
    case ASCII:
    case TEXT:
    case VARCHAR:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, String.class);
      break;
    case BOOLEAN:
      getter = PojoUtils.createGetterBoolean(tuplePayloadClass, getterExpr);
      break;
    case FLOAT:
      getter = PojoUtils.createGetterFloat(tuplePayloadClass, getterExpr);
      break;
    case DOUBLE:
      getter = PojoUtils.createGetterDouble(tuplePayloadClass, getterExpr);
      break;
    case DECIMAL:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, BigDecimal.class);
      break;
    case SET:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Set.class);
      break;
    case MAP:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Map.class);
      break;
    case LIST:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, List.class);
      break;
    case TIMESTAMP:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Date.class);
      break;
    case UUID:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, UUID.class);
      break;
    case UDT:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, userDefinedTypesClass.get(getterExpr));
      break;
    default:
      getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Object.class);
      break;
  }
  return getter;
}
 

void checkType(int i, DataType.Name name) {
    DataType defined = getType(i);
    if (name != defined.getName())
        throw new InvalidTypeException(String.format("Column %s is of type %s", getName(i), defined));
}
 

public static Class<?> asJavaClass(DataType dataType) {
    DataType.Name name = dataType.getName();
    return asJavaClass(name);
}
 

private DataType.Name validateName(DataType dataType) {
    if (DataTypeHelper.isNumber(dataType)) {
        return dataType.getName();
    }
    throw new IllegalArgumentException("DataType " + dataType + "is not a number");
}
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();

        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();

        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }