coral-beam-runtime/src/main/java/com/linkedin/beam/excution/BeamExecUtil.java
public static GenericArray toAvroArray(Object obj, Schema schema) {
    if (obj == null) {
      return null;
    }

    if (obj instanceof GenericArray) {
      return (GenericArray) obj;
    }

    if (obj instanceof DataBag) {
      final DataBag bag = (DataBag) obj;
      final GenericData.Array<Object> array = new GenericData.Array<>(new Long(bag.size()).intValue(), schema);
      for (Tuple tuple : bag) {
        try {
          if (schema.getElementType() != null && schema.getElementType().getType() == Schema.Type.RECORD) {
            array.add(toAvroRecord(tuple, schema.getElementType()));
          } else if (tuple.size() == 1) {
            array.add(tuple.get(0));
          } else {
            throw new RuntimeException("Can't pack " + tuple + " into schema " + schema);
          }
        } catch (ExecException e) {
          throw new RuntimeException("Can't pack " + tuple + " into schema " + schema, e);
        }
      }
      return array;
    }

    throw new RuntimeException("Cannot convert " + obj.getClass().getName() + " object into GenericArray");
  }
/**
   * Converts an object, which can be already an Avro GenericArray or a Pig DataBag,
   * into an Avro GenericArray. Returns null if object is null.
   */
Converts an object, which can be already an Avro GenericArray or a Pig DataBag,
into an Avro GenericArray. Returns null if object is null.
Converts an object which can be already an Avro GenericArray or a Pig DataBag into an Avro GenericArray . Returns null if object is null .
public static GenericArray toAvroArray(Object obj, Schema schema) {
    if (obj == null) {
      return null;
    }
    if (obj instanceof GenericArray) {
      return (GenericArray) obj;
    }
    if (obj instanceof DataBag) {
      final DataBag bag = (DataBag) obj;
      final GenericData.Array<Object> array = new GenericData.Array<>(new Long(bag.size()).intValue(), schema);
      for (Tuple tuple : bag) {
        try {
          if (schema.getElementType() != null && schema.getElementType().getType() == Schema.Type.RECORD) {
            array.add(toAvroRecord(tuple, schema.getElementType()));
          } else if (tuple.size() == 1) {
            array.add(tuple.get(0));
          } else {
            throw new RuntimeException("Can't pack " + tuple + " into schema " + schema);
          }
        } catch (ExecException e) {
          throw new RuntimeException("Can't pack " + tuple + " into schema " + schema, e);
        }
      }
      return array;
    }
    throw new RuntimeException("Cannot convert " + obj.getClass().getName() + " object into GenericArray");
  }
public static GenericArray toAvroArray ( Object obj , Schema schema ) { if ( obj == null ) { return null ; } if ( obj instanceof GenericArray ) { return ( GenericArray ) obj ; } if ( obj instanceof DataBag ) { final DataBag bag = ( DataBag ) obj ; final GenericData . Array < Object > array = new GenericData . Array < > ( new Long ( bag . size ( ) ) . intValue ( ) , schema ) ; for ( Tuple tuple : bag ) { try { if ( schema . getElementType ( ) != null && schema . getElementType ( ) . getType ( ) == Schema . Type . RECORD ) { array . add ( toAvroRecord ( tuple , schema . getElementType ( ) ) ) ; } else if ( tuple . size ( ) == 1 ) { array . add ( tuple . get ( 0 ) ) ; } else { throw new RuntimeException ( "Can't pack " + tuple + " into schema " + schema ) ; } } catch ( ExecException e ) { throw new RuntimeException ( "Can't pack " + tuple + " into schema " + schema , e ) ; } } return array ; } throw new RuntimeException ( "Cannot convert " + obj . getClass ( ) . getName ( ) + " object into GenericArray" ) ; }
Converts an object, which can be already an Avro GenericArray or a Pig DataBag,
 into an Avro GenericArray.
Converts an object which can be already an Avro GenericArray or a Pig DataBag into an Avro GenericArray .