public Mutation serialize(Object obj, ObjectInspector objInspector) throws SerDeException,
      IOException {
    if (objInspector.getCategory() != ObjectInspector.Category.STRUCT) {
      throw new SerDeException(getClass().toString()
          + " can only serialize struct types, but we got: " + objInspector.getTypeName());
    }

    // Prepare the field ObjectInspectors
    StructObjectInspector soi = (StructObjectInspector) objInspector;
    List<? extends StructField> fields = soi.getAllStructFieldRefs();
    List<Object> columnValues = soi.getStructFieldsDataAsList(obj);

    // Fail if we try to access an offset out of bounds
    if (rowIdOffset >= fields.size()) {
      throw new IllegalStateException(
          "Attempted to access field outside of definition for struct. Have " + fields.size()
              + " fields and tried to access offset " + rowIdOffset);
    }

    StructField field = fields.get(rowIdOffset);
    Object value = columnValues.get(rowIdOffset);

    // The ObjectInspector for the row ID
    ObjectInspector fieldObjectInspector = field.getFieldObjectInspector();

    // Serialize the row component using the RowIdFactory. In the normal case, this will just
    // delegate back to the "local" serializeRowId method
    byte[] data = rowIdFactory.serializeRowId(value, field, output);

    // Set that as the row id in the mutation
    Mutation mutation = new Mutation(data);

    // Each column in the row
    for (int i = 0; i < fields.size(); i++) {
      if (rowIdOffset == i) {
        continue;
      }

      // Get the relevant information for this column
      field = fields.get(i);
      value = columnValues.get(i);

      // Despite having a fixed schema from Hive, we have sparse columns in Accumulo
      if (null == value) {
        continue;
      }

      // The ObjectInspector for the current column
      fieldObjectInspector = field.getFieldObjectInspector();

      // Make sure we got the right implementation of a ColumnMapping
      ColumnMapping mapping = mappings.get(i);
      if (mapping instanceof HiveAccumuloColumnMapping) {
        serializeColumnMapping((HiveAccumuloColumnMapping) mapping, fieldObjectInspector, value,
            mutation);
      } else if (mapping instanceof HiveAccumuloMapColumnMapping) {
        serializeColumnMapping((HiveAccumuloMapColumnMapping) mapping, fieldObjectInspector, value,
            mutation);
      } else {
        throw new IllegalArgumentException("Mapping for " + field.getFieldName()
            + " was not a HiveColumnMapping, but was " + mapping.getClass());
      }

    }

    return mutation;
  }
Serializes a Struct to an Accumulo row as per the definition provided by the
{@link ColumnMapping}s
accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/serde/AccumuloRowSerializer.java