coral-beam-runtime/src/main/java/com/linkedin/beam/excution/BeamArrayFlatten.java
public static Collection<GenericRecord> flatten(GenericRecord inputRecord, List<String> flattenCols,
      Schema outputSchema) {
    List<GenericRecord> results = new ArrayList<>();
    if (inputRecord == null) {
      return results;
    }
    Schema inputSchema = inputRecord.getSchema();
    Map<String, Object> partialRecord = new HashMap<>();
    for (Schema.Field field : inputSchema.getFields()) {
      partialRecord.put(field.name(), inputRecord.get(field.name()));
    }
    List<Map<String, Object>> partialResults = new ArrayList<>();
    partialResults.add(partialRecord);

    for (String flattenCol : flattenCols) {
      Schema flattenSchema = getFlattenElementSchema(inputSchema, flattenCol);
      if (flattenSchema == null) {
        throw new RuntimeException("Column " + flattenCol + " is not an array in records"
            + " with schema: " + inputSchema);
      }
      Object obj = inputRecord.get(flattenCol);
      List<Object> arrayValue = null;
      if (obj != null) {
        if (obj.getClass().isArray()) {
          arrayValue = Arrays.asList((Object[]) obj);
        } else if (obj instanceof Collection) {
          if (obj instanceof List) {
            arrayValue = (List<Object>) obj;
          } else {
            arrayValue = new ArrayList<>((Collection<Object>) obj);
          }
        } else {
          throw new RuntimeException("Invalid avro array value. Value type: " + obj.getClass().getName());
        }
      }
      partialResults = extendFlattenRecords(partialResults, flattenCol, arrayValue, flattenSchema);
    }

    for (Map<String, Object> outputRec : partialResults) {
      GenericRecord record = new GenericData.Record(outputSchema);
      for (Schema.Field field : outputSchema.getFields()) {
        record.put(field.name(), outputRec.get(field.name()));
      }
      results.add(record);
    }
    return results;
  }
/**
   * Flattens a record on a given of array columns. These columns must have ARRAY schema type
   * and the input values for those column must be either Java array or Collection.
   *
   * @param inputRecord Input Avro records
   * @param flattenCols Columns to be flattened
   * @param outputSchema Schemas of ouput records
   * @return Collection of output records.
   */
Flattens a record on a given of array columns. These columns must have ARRAY schema type
and the input values for those column must be either Java array or Collection.

@param inputRecord Input Avro records
@param flattenCols Columns to be flattened
@param outputSchema Schemas of ouput records
@return Collection of output records.
Flattens a record on a given of array columns . These columns must have ARRAY schema type and the input values for those column must be either Java array or Collection . @param inputRecord Input Avro records @param flattenCols Columns to be flattened @param outputSchema Schemas of ouput records @return Collection of output records .
public static Collection<GenericRecord> flatten(GenericRecord inputRecord, List<String> flattenCols,
      Schema outputSchema) {
    List<GenericRecord> results = new ArrayList<>();
    if (inputRecord == null) {
      return results;
    }
    Schema inputSchema = inputRecord.getSchema();
    Map<String, Object> partialRecord = new HashMap<>();
    for (Schema.Field field : inputSchema.getFields()) {
      partialRecord.put(field.name(), inputRecord.get(field.name()));
    }
    List<Map<String, Object>> partialResults = new ArrayList<>();
    partialResults.add(partialRecord);
    for (String flattenCol : flattenCols) {
      Schema flattenSchema = getFlattenElementSchema(inputSchema, flattenCol);
      if (flattenSchema == null) {
        throw new RuntimeException("Column " + flattenCol + " is not an array in records"
            + " with schema: " + inputSchema);
      }
      Object obj = inputRecord.get(flattenCol);
      List<Object> arrayValue = null;
      if (obj != null) {
        if (obj.getClass().isArray()) {
          arrayValue = Arrays.asList((Object[]) obj);
        } else if (obj instanceof Collection) {
          if (obj instanceof List) {
            arrayValue = (List<Object>) obj;
          } else {
            arrayValue = new ArrayList<>((Collection<Object>) obj);
          }
        } else {
          throw new RuntimeException("Invalid avro array value. Value type: " + obj.getClass().getName());
        }
      }
      partialResults = extendFlattenRecords(partialResults, flattenCol, arrayValue, flattenSchema);
    }
    for (Map<String, Object> outputRec : partialResults) {
      GenericRecord record = new GenericData.Record(outputSchema);
      for (Schema.Field field : outputSchema.getFields()) {
        record.put(field.name(), outputRec.get(field.name()));
      }
      results.add(record);
    }
    return results;
  }
public static Collection < GenericRecord > flatten ( GenericRecord inputRecord , List < String > flattenCols , Schema outputSchema ) { List < GenericRecord > results = new ArrayList < > ( ) ; if ( inputRecord == null ) { return results ; } Schema inputSchema = inputRecord . getSchema ( ) ; Map < String , Object > partialRecord = new HashMap < > ( ) ; for ( Schema . Field field : inputSchema . getFields ( ) ) { partialRecord . put ( field . name ( ) , inputRecord . get ( field . name ( ) ) ) ; } List < Map < String , Object > > partialResults = new ArrayList < > ( ) ; partialResults . add ( partialRecord ) ; for ( String flattenCol : flattenCols ) { Schema flattenSchema = getFlattenElementSchema ( inputSchema , flattenCol ) ; if ( flattenSchema == null ) { throw new RuntimeException ( "Column " + flattenCol + " is not an array in records" + " with schema: " + inputSchema ) ; } Object obj = inputRecord . get ( flattenCol ) ; List < Object > arrayValue = null ; if ( obj != null ) { if ( obj . getClass ( ) . isArray ( ) ) { arrayValue = Arrays . asList ( ( Object [ ] ) obj ) ; } else if ( obj instanceof Collection ) { if ( obj instanceof List ) { arrayValue = ( List < Object > ) obj ; } else { arrayValue = new ArrayList < > ( ( Collection < Object > ) obj ) ; } } else { throw new RuntimeException ( "Invalid avro array value. Value type: " + obj . getClass ( ) . getName ( ) ) ; } } partialResults = extendFlattenRecords ( partialResults , flattenCol , arrayValue , flattenSchema ) ; } for ( Map < String , Object > outputRec : partialResults ) { GenericRecord record = new GenericData . Record ( outputSchema ) ; for ( Schema . Field field : outputSchema . getFields ( ) ) { record . put ( field . name ( ) , outputRec . get ( field . name ( ) ) ) ; } results . add ( record ) ; } return results ; }
Flattens a record on a given of array columns.
Flattens a record on a given of array columns .