@SuppressWarnings({ "unchecked", "rawtypes" })
  public Record toRecord(ITuple tuple, Record reuse) throws IOException {
		Record record = reuse;
		if (record == null){
			record = new Record(avroSchema);
		}
		if (schemaValidation && !tuple.getSchema().equals(pangoolSchema)){
			throw new IOException("Tuple '"+tuple + "' " +
					"contains schema not expected." +
					"Expected schema '"+ pangoolSchema + " and actual: " + tuple.getSchema());
		}
		for(int i = 0; i < pangoolSchema.getFields().size(); i++) {
			Object obj = tuple.get(i);
			Field field = pangoolSchema.getField(i);
			if (obj == null){
				throw new IOException("Field '" 
			+ field.getName() + "' can't be null in tuple:" + tuple);
			}
			
			switch(field.getType()){
			case INT:
			case LONG:
			case FLOAT:
			case BOOLEAN:
			case DOUBLE:
			case BYTES:
				record.put(i, obj); //optimistic
				break;
			case OBJECT:
				Serializer customSer = customSerializers[i];
				DataOutputBuffer buffer = buffers[i];
				buffer.reset();
				if (customSer != null){
					customSer.open(buffer);
					customSer.serialize(obj);
					customSer.close(); //TODO is this safe ?
				} else {
					hadoopSer.ser(obj, buffer);
				}
				//TODO this byteBuffer instances should be cached and reused
				ByteBuffer byteBuffer = ByteBuffer.wrap(buffer.getData(), 0,buffer.getLength());
				record.put(i, byteBuffer);
				break;
			case ENUM:
				record.put(i,obj.toString());
				break;
			case STRING:
				record.put(i,new Utf8(obj.toString())); //could be directly String ?
				break;
			default:
					throw 
					new IOException("Not correspondence to Avro type from Pangool type " + field.getType());
			}
		}
		return record;
	}
@ SuppressWarnings ( { "unchecked" , "rawtypes" } ) public Record toRecord ( ITuple tuple , Record reuse ) throws IOException { Record record = reuse ; if ( record == null ) { record = new Record ( avroSchema ) ; } if ( schemaValidation && ! tuple . getSchema ( ) . equals ( pangoolSchema ) ) { throw new IOException ( "Tuple '" + tuple + "' " + "contains schema not expected." + "Expected schema '" + pangoolSchema + " and actual: " + tuple . getSchema ( ) ) ; } for ( int i = 0 ; i < pangoolSchema . getFields ( ) . size ( ) ; i ++ ) { Object obj = tuple . get ( i ) ; Field field = pangoolSchema . getField ( i ) ; if ( obj == null ) { throw new IOException ( "Field '" + field . getName ( ) + "' can't be null in tuple:" + tuple ) ; } switch ( field . getType ( ) ) { case INT : case LONG : case FLOAT : case BOOLEAN : case DOUBLE : case BYTES : record . put ( i , obj ) ; break ; case OBJECT : Serializer customSer = customSerializers [ i ] ; DataOutputBuffer buffer = buffers [ i ] ; buffer . reset ( ) ; if ( customSer != null ) { customSer . open ( buffer ) ; customSer . serialize ( obj ) ; customSer . close ( ) ; } else { hadoopSer . ser ( obj , buffer ) ; } ByteBuffer byteBuffer = ByteBuffer . wrap ( buffer . getData ( ) , 0 , buffer . getLength ( ) ) ; record . put ( i , byteBuffer ) ; break ; case ENUM : record . put ( i , obj . toString ( ) ) ; break ; case STRING : record . put ( i , new Utf8 ( obj . toString ( ) ) ) ; break ; default : throw new IOException ( "Not correspondence to Avro type from Pangool type " + field . getType ( ) ) ; } } return record ; }
Moves data between a Tuple and an Avro Record