@Override
  public KuduWritable serialize(Object obj, ObjectInspector objectInspector) throws SerDeException {
    Preconditions.checkArgument(objectInspector.getCategory() == Category.STRUCT);

    StructObjectInspector soi = (StructObjectInspector) objectInspector;
    List<Object> writableObj = soi.getStructFieldsDataAsList(obj);
    List<? extends StructField> fields = soi.getAllStructFieldRefs();

    PartialRow row = schema.newPartialRow();
    for (int i = 0; i < schema.getColumnCount(); i++) {
      StructField field = fields.get(i);
      Object value = writableObj.get(i);

      if (value == null) {
        row.setNull(i);
      } else {
        Type type = schema.getColumnByIndex(i).getType();
        ObjectInspector inspector = field.getFieldObjectInspector();
        switch (type) {
        case BOOL:
          boolean boolVal = ((BooleanObjectInspector) inspector).get(value);
          row.addBoolean(i, boolVal);
          break;
        case INT8:
          byte byteVal = ((ByteObjectInspector) inspector).get(value);
          row.addByte(i, byteVal);
          break;
        case INT16:
          short shortVal = ((ShortObjectInspector) inspector).get(value);
          row.addShort(i, shortVal);
          break;
        case INT32:
          int intVal = ((IntObjectInspector) inspector).get(value);
          row.addInt(i, intVal);
          break;
        case INT64:
          long longVal = ((LongObjectInspector) inspector).get(value);
          row.addLong(i, longVal);
          break;
        case UNIXTIME_MICROS:
          // Calling toSqlTimestamp and using the addTimestamp API ensures we properly
          // convert Hive localDateTime to UTC.
          java.sql.Timestamp timestampVal = ((TimestampObjectInspector) inspector)
              .getPrimitiveJavaObject(value).toSqlTimestamp();
          row.addTimestamp(i, timestampVal);
          break;
        case DECIMAL:
          HiveDecimal decimalVal = ((HiveDecimalObjectInspector) inspector)
              .getPrimitiveJavaObject(value);
          row.addDecimal(i, decimalVal.bigDecimalValue());
          break;
        case FLOAT:
          float floatVal = ((FloatObjectInspector) inspector).get(value);
          row.addFloat(i, floatVal);
          break;
        case DOUBLE:
          double doubleVal = ((DoubleObjectInspector) inspector).get(value);
          row.addDouble(i, doubleVal);
          break;
        case STRING:
          String stringVal = ((StringObjectInspector) inspector).getPrimitiveJavaObject(value);
          row.addString(i, stringVal);
          break;
        case BINARY:
          byte[] bytesVal = ((BinaryObjectInspector) inspector).getPrimitiveJavaObject(value);
          row.addBinary(i, bytesVal);
          break;
        default:
          throw new SerDeException("Unsupported column type: " + type.name());
        }
      }
    }
    return new KuduWritable(row);
  }
Serialize an object by navigating inside the Object with the ObjectInspector.
kudu-handler/src/java/org/apache/hadoop/hive/kudu/KuduSerDe.java