import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Returns a clone with exactly the requested fields shallowly copied
 */
private static<T extends Persistent> T getPersistent(T obj, String[] fields) {
  List<Field> otherFields = obj.getSchema().getFields();
  String[] otherFieldStrings = new String[otherFields.size()];
  for(int i = 0; i<otherFields.size(); i++ ){
    otherFieldStrings[i] = otherFields.get(i).name();
  }
  if(Arrays.equals(fields, otherFieldStrings)) { 
    return obj;
  }
  T newObj = AvroUtils.deepClonePersistent(obj);
  for (Field otherField : otherFields) {
    int index = otherField.pos();
    newObj.put(index, obj.get(index));
  }
  return newObj;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Sets the output parameters for the conf that Spark will use
 *
 * @param job             the job to set the properties for
 * @param dataStoreClass  the datastore class
 * @param keyClass        output key class
 * @param persistentClass output value class
 */
@SuppressWarnings("rawtypes")
public <K, V extends Persistent> Configuration generateOutputConf(Job job,
    Class<? extends DataStore> dataStoreClass,
    Class<K> keyClass, Class<V> persistentClass) {

  job.setOutputFormatClass(GoraOutputFormat.class);
  job.setOutputKeyClass(keyClass);
  job.setOutputValueClass(persistentClass);

  job.getConfiguration().setClass(GoraOutputFormat.DATA_STORE_CLASS, dataStoreClass,
          DataStore.class);
  job.getConfiguration().setClass(GoraOutputFormat.OUTPUT_KEY_CLASS, keyClass, Object.class);
  job.getConfiguration().setClass(GoraOutputFormat.OUTPUT_VALUE_CLASS,
          persistentClass, Persistent.class);
  return job.getConfiguration();
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Constructor
 * @param keyClass
 * @param persistentClass
 */
public BeanFactoryWSImpl(Class<K> keyClass, Class<T> persistentClass) {
  this.keyClass = keyClass;
  this.persistentClass = persistentClass;
  
  try {
    if(ReflectionUtils.hasConstructor(keyClass)) {
      this.keyConstructor = ReflectionUtils.getConstructor(keyClass);
      this.key = keyConstructor.newInstance(ReflectionUtils.EMPTY_OBJECT_ARRAY);
    }
    this.persistent = ReflectionUtils.newInstance(persistentClass);
  } catch (Exception ex) {
    throw new RuntimeException(ex);
  }
  
  isKeyPersistent = Persistent.class.isAssignableFrom(keyClass);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Default constructor for this class.
 * @param keyClass.
 * @param persistentClass
 */
public BeanFactoryImpl(Class<K> keyClass, Class<T> persistentClass) {
  this.keyClass = keyClass;
  this.persistentClass = persistentClass;
  
  try {
    if(ReflectionUtils.hasConstructor(keyClass)) {
      this.keyConstructor = ReflectionUtils.getConstructor(keyClass);
      this.key = keyConstructor.newInstance(ReflectionUtils.EMPTY_OBJECT_ARRAY);
    }
    this.persistent = ReflectionUtils.newInstance(persistentClass);
  } catch (Exception ex) {
    throw new RuntimeException(ex);
  }
  
  isKeyPersistent = Persistent.class.isAssignableFrom(keyClass);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Sets the output parameters for the job 
 * @param job the job to set the properties for
 * @param dataStoreClass the datastore class
 * @param keyClass output key class
 * @param persistentClass output value class
 * @param reuseObjects whether to reuse objects in serialization
 */
@SuppressWarnings("rawtypes")
public static <K, V extends Persistent> void setOutput(Job job,
    Class<? extends DataStore> dataStoreClass,
    Class<K> keyClass, Class<V> persistentClass,
    boolean reuseObjects) {

  Configuration conf = job.getConfiguration();

  GoraMapReduceUtils.setIOSerializations(conf, reuseObjects);

  job.setOutputFormatClass(GoraOutputFormat.class);
  job.setOutputKeyClass(keyClass);
  job.setOutputValueClass(persistentClass);
  conf.setClass(GoraOutputFormat.DATA_STORE_CLASS, dataStoreClass,
      DataStore.class);
  conf.setClass(GoraOutputFormat.OUTPUT_KEY_CLASS, keyClass, Object.class);
  conf.setClass(GoraOutputFormat.OUTPUT_VALUE_CLASS,
      persistentClass, Persistent.class);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Initializes the Mapper, and sets input parameters for the job. All of 
 * the records in the dataStore are used as the input. If you want to 
 * include a specific subset, use one of the overloaded methods which takes
 * query parameter.
 * @param job the job to set the properties for
 * @param dataStoreClass the datastore class
 * @param inKeyClass Map input key class
 * @param inValueClass Map input value class
 * @param outKeyClass Map output key class
 * @param outValueClass Map output value class
 * @param mapperClass the mapper class extending GoraMapper
 * @param partitionerClass optional partitioner class
 * @param reuseObjects whether to reuse objects in serialization
 */
@SuppressWarnings("rawtypes")
public static <K1, V1 extends Persistent, K2, V2> void initMapperJob(
    Job job,
    Class<? extends DataStore<K1,V1>> dataStoreClass,
    Class<K1> inKeyClass, 
    Class<V1> inValueClass,
    Class<K2> outKeyClass, 
    Class<V2> outValueClass,
    Class<? extends GoraMapper> mapperClass,
    Class<? extends Partitioner> partitionerClass, 
    boolean reuseObjects) throws IOException {
  
  //set the input via GoraInputFormat
  GoraInputFormat.setInput(job, dataStoreClass, inKeyClass, inValueClass, reuseObjects);

  job.setMapperClass(mapperClass);
  job.setMapOutputKeyClass(outKeyClass);
  job.setMapOutputValueClass(outValueClass);

  if (partitionerClass != null) {
    job.setPartitionerClass(partitionerClass);
  }
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
protected String[] getFields() {
  List<Field> schemaFields = beanFactory.getCachedPersistent().getSchema().getFields();
  
  List<Field> list = new ArrayList<>();
  for (Field field : schemaFields) {
    if (!Persistent.DIRTY_BYTES_FIELD_NAME.equalsIgnoreCase(field.name())) {
      list.add(field);
    }
  }
  schemaFields = list;
  
  String[] fieldNames = new String[schemaFields.size()];
  for(int i = 0; i<fieldNames.length; i++ ){
    fieldNames[i] = schemaFields.get(i).name();
  }
  
  return fieldNames;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
public static String[] getFields(List<Field> schemaFields) {
  
  List<Field> list = new ArrayList<>();
  for (Field field : schemaFields) {
    if (!Persistent.DIRTY_BYTES_FIELD_NAME.equalsIgnoreCase(field.name())) {
      list.add(field);
    }
  }
  schemaFields = list;
  
  String[] fieldNames = new String[schemaFields.size()];
  for(int i = 0; i<fieldNames.length; i++ ){
    fieldNames[i] = schemaFields.get(i).name();
  }
  
  return fieldNames;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
@Test
public void shouldInvokeDastorePut() throws Exception {

    when(mockCamelExchange.getIn()).thenReturn(mockCamelMessage);
    when(mockCamelMessage.getHeader(GoraAttribute.GORA_OPERATION.value)).thenReturn("PUT");

    final Long sampleKey = new Long(2);
    when(mockCamelMessage.getHeader(GoraAttribute.GORA_KEY.value)).thenReturn(sampleKey);

    final Persistent sampleValue = mock(Persistent.class);
    when(mockCamelMessage.getBody(Persistent.class)).thenReturn(sampleValue);

    final Message outMessage = mock(Message.class);
    when(mockCamelExchange.getOut()).thenReturn(outMessage);

    final GoraProducer producer = new GoraProducer(mockGoraEndpoint, mockGoraConfiguration, mockDatastore);
    producer.process(mockCamelExchange);

    verify(mockCamelExchange, atLeastOnce()).getIn();
    verify(mockCamelMessage, atLeastOnce()).getHeader(GoraAttribute.GORA_OPERATION.value);
    verify(mockCamelMessage, atLeastOnce()).getHeader(GoraAttribute.GORA_KEY.value);
    verify(mockCamelMessage, atLeastOnce()).getBody(Persistent.class);
    verify(mockDatastore, atMost(1)).put(sampleKey, sampleValue);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static <K, T extends Persistent> IDynamoDB<K, T> buildDynamoDBStore(
    DynamoDBUtils.DynamoDBType serType) {
  final IDynamoDB<K, T> ds;
  switch (serType) {
    case DYNAMO:
      ds = new DynamoDBNativeStore<K, T>();
      LOG.debug("Using DynamoDB based serialization mode.");
      break;
    case AVRO:
      ds = (IDynamoDB<K, T>) new DynamoDBAvroStore<K, PersistentBase>();
      LOG.debug("Using Avro based serialization mode.");
      break;
    default:
      throw new IllegalStateException("Serialization mode not supported.");
  }
  return ds;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Initializes the Mapper, and sets input parameters for the job
 * @param job the job to set the properties for
 * @param query the query to get the inputs from
 * @param dataStore the datastore as the input
 * @param outKeyClass Map output key class
 * @param outValueClass Map output value class
 * @param mapperClass the mapper class extending GoraMapper
 * @param partitionerClass optional partitioner class
 * @param reuseObjects whether to reuse objects in serialization
 */
@SuppressWarnings("rawtypes")
public static <K1, V1 extends Persistent, K2, V2> void initMapperJob(
    Job job, 
    Query<K1,V1> query,
    DataStore<K1,V1> dataStore,
    Class<K2> outKeyClass, 
    Class<V2> outValueClass,
    Class<? extends GoraMapper> mapperClass,
    Class<? extends Partitioner> partitionerClass, 
    boolean reuseObjects) throws IOException {
  //set the input via GoraInputFormat
  GoraInputFormat.setInput(job, query, dataStore, reuseObjects);

  job.setMapperClass(mapperClass);
  job.setMapOutputKeyClass(outKeyClass);
  job.setMapOutputValueClass(outValueClass);

  if (partitionerClass != null) {
    job.setPartitionerClass(partitionerClass);
  }
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
* Sets the output parameters for the conf that Spark will use
*
* @param job             the job to set the properties for
* @param dataStoreClass  the datastore class
* @param keyClass        output key class
* @param persistentClass output value class
* @return a populated output {@link org.apache.hadoop.conf.Configuration} 
*/
 @SuppressWarnings("rawtypes")
 public <K, V extends Persistent> Configuration generateOutputConf(Job job,
     Class<? extends DataStore> dataStoreClass,
     Class<K> keyClass, Class<V> persistentClass) {

   job.setOutputFormatClass(GoraOutputFormat.class);
   job.setOutputKeyClass(keyClass);
   job.setOutputValueClass(persistentClass);

   job.getConfiguration().setClass(GoraOutputFormat.DATA_STORE_CLASS, dataStoreClass,
           DataStore.class);
   job.getConfiguration().setClass(GoraOutputFormat.OUTPUT_KEY_CLASS, keyClass, Object.class);
   job.getConfiguration().setClass(GoraOutputFormat.OUTPUT_VALUE_CLASS,
           persistentClass, Persistent.class);
   return job.getConfiguration();
 }
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Default constructor for this class.
 *
 * @param keyClass class of the keys
 * @param persistentClass class of the [{@link Persistent} objects to be stored
 */
public BeanFactoryImpl(Class<K> keyClass, Class<T> persistentClass) {
  this.keyClass = keyClass;
  this.persistentClass = persistentClass;
  
  try {
    if(ReflectionUtils.hasConstructor(keyClass)) {
      this.keyConstructor = ReflectionUtils.getConstructor(keyClass);
      this.key = keyConstructor.newInstance(ReflectionUtils.EMPTY_OBJECT_ARRAY);
    }
    this.persistent = ReflectionUtils.newInstance(persistentClass);
  } catch (Exception ex) {
    throw new RuntimeException(ex);
  }
  
  isKeyPersistent = Persistent.class.isAssignableFrom(keyClass);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Sets the output parameters for the job
 *
 * @param job             the job to set the properties for
 * @param dataStoreClass  the datastore class
 * @param keyClass        output key class
 * @param persistentClass output value class
 * @param reuseObjects    whether to reuse objects in serialization
 * @param <K> key class for the {@link org.apache.gora.store.DataStore}
 * @param <V> value class for the {@link org.apache.gora.store.DataStore}
 */
@SuppressWarnings("rawtypes")
public static <K, V extends Persistent> void setOutput(Job job,
    Class<? extends DataStore> dataStoreClass,
    Class<K> keyClass, Class<V> persistentClass,
    boolean reuseObjects) {

  Configuration conf = job.getConfiguration();

  GoraMapReduceUtils.setIOSerializations(conf, reuseObjects);

  job.setOutputFormatClass(GoraOutputFormat.class);
  job.setOutputKeyClass(keyClass);
  job.setOutputValueClass(persistentClass);
  conf.setClass(GoraOutputFormat.DATA_STORE_CLASS, dataStoreClass,
      DataStore.class);
  conf.setClass(GoraOutputFormat.OUTPUT_KEY_CLASS, keyClass, Object.class);
  conf.setClass(GoraOutputFormat.OUTPUT_VALUE_CLASS,
      persistentClass, Persistent.class);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Initializes the Mapper, and sets input parameters for the job
 *
 * @param job              the job to set the properties for
 * @param query            the query to get the inputs from
 * @param outKeyClass      Map output key class
 * @param outValueClass    Map output value class
 * @param mapperClass      the mapper class extending GoraMapper
 * @param partitionerClass optional partitioner class
 * @param reuseObjects     whether to reuse objects in serialization
 * @param <K1> Map input key class
 * @param <V1> Map input value class
 * @param <K2> Map output key class
 * @param <V2> Map output value class
 * @throws IOException if there is an error initializing the Map job
 */
@SuppressWarnings("rawtypes")
public static <K1, V1 extends Persistent, K2, V2> void initMapperJob(
    Job job, 
    Query<K1,V1> query,
    Class<K2> outKeyClass,
    Class<V2> outValueClass,
    Class<? extends GoraMapper> mapperClass,
    Class<? extends Partitioner> partitionerClass, 
    boolean reuseObjects) throws IOException {
  //set the input via GoraInputFormat
  GoraInputFormat.setInput(job, query, reuseObjects);

  job.setMapperClass(mapperClass);
  job.setMapOutputKeyClass(outKeyClass);
  job.setMapOutputValueClass(outValueClass);

  if (partitionerClass != null) {
    job.setPartitionerClass(partitionerClass);
  }
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * Instantiate <i>the default</i> {@link DataStore} wrapped over caching dataStore which provides caching
 * abstraction over the GORA persistence dataStore.
 * Uses default properties. Uses 'null' schema.
 *
 * Note:
 *    consider that default dataStore is always visible
 *
 * @param keyClass The key class.
 * @param persistent The value class.
 * @param conf {@link Configuration} To be used be the store.
 * @param isCacheEnabled Caching enable or not.
 * @return A new store instance.
 * @throws GoraException If cache or persistency dataStore initialization interrupted.
 */
@SuppressWarnings("unchecked")
public static <K, T extends Persistent> DataStore<K, T> getDataStore(
        Class<K> keyClass, Class<T> persistent, Configuration conf, boolean isCacheEnabled) throws GoraException {
  Properties createProps = createProps();
  Class<? extends DataStore<K, T>> c;
  try {
    if (isCacheEnabled) {
      c = (Class<? extends DataStore<K, T>>) Class.forName(getDefaultCacheDataStore(createProps));
    } else {
      c = (Class<? extends DataStore<K, T>>) Class.forName(getDefaultDataStore(createProps));
    }
  } catch (Exception ex) {
    throw new GoraException(ex);
  }
  return createDataStore(c, keyClass, persistent, conf, createProps, null);
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
protected String[] getFields() {
  List<Field> schemaFields = beanFactory.getCachedPersistent().getSchema().getFields();

  List<Field> list = new ArrayList<>();
  for (Field field : schemaFields) {
    if (!Persistent.DIRTY_BYTES_FIELD_NAME.equalsIgnoreCase(field.name())) {
      list.add(field);
    }
  }
  schemaFields = list;

  String[] fieldNames = new String[schemaFields.size()];
  for(int i = 0; i<fieldNames.length; i++ ){
    fieldNames[i] = schemaFields.get(i).name();
  }

  return fieldNames;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @param key
 * @param fields
 * @return
 */
@Override
public Persistent get(Object key, String[] fields) {
  if (fields == null) {
    fields = getFields();
  }
  String cqlQuery = CassandraQueryFactory.getSelectObjectWithFieldsQuery(mapping, fields);
  SimpleStatement statement = new SimpleStatement(cqlQuery, key);
  if (readConsistencyLevel != null) {
    statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel));
  }
  ResultSet results = client.getSession().execute(statement);
  Result<T> objects = mapper.map(results);
  List<T> objectList = objects.all();
  if (objectList != null) {
    LOG.debug("Object is found for key : {}", key);
    return objectList.get(0);
  }
  LOG.debug("Object is not found for key : {}", key);
  return null;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @param key
 * @param fields
 * @return
 */
@Override
public Persistent get(Object key, String[] fields) {
  if (fields == null) {
    fields = getFields();
  }
  ArrayList<String> cassandraKeys = new ArrayList<>();
  ArrayList<Object> cassandraValues = new ArrayList<>();
  AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues);
  String cqlQuery = CassandraQueryFactory.getSelectObjectWithFieldsQuery(mapping, fields, cassandraKeys);
  SimpleStatement statement = new SimpleStatement(cqlQuery, cassandraValues.toArray());
  if (readConsistencyLevel != null) {
    statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel));
  }
  ResultSet resultSet = this.client.getSession().execute(statement);
  Iterator<Row> iterator = resultSet.iterator();
  ColumnDefinitions definitions = resultSet.getColumnDefinitions();
  T obj = null;
  if (iterator.hasNext()) {
    obj = cassandraDataStore.newPersistent();
    AbstractGettableData row = (AbstractGettableData) iterator.next();
    populateValuesToPersistent(row, definitions, obj, fields);
  }
  return obj;
}
 

import org.apache.gora.persistency.Persistent; //导入依赖的package包/类
public Persistent clone(Persistent persistent, Schema schema) {
  Persistent cloned = (PersistentBase)persistent.newInstance(new StateManagerImpl());
  List<Field> fields = schema.getFields();
  for(Field field: fields) {
    int pos = field.pos();
    switch(field.schema().getType()) {
      case MAP    :
      case ARRAY  :
      case RECORD : 
      case STRING : ((PersistentBase)cloned).put(pos, cloneObject(
          field.schema(), ((PersistentBase)persistent).get(pos), ((PersistentBase)cloned).get(pos))); break;
      case NULL   : break;
      default     : ((PersistentBase)cloned).put(pos, ((PersistentBase)persistent).get(pos)); break;
    }
  }
  
  return cloned;
}