code: public static <E> PCollection<E> partition(PCollection<E> collection,
                                               View<E> view,
                                               int numWriters, int numPartitionWriters) {
      //ensure the number of writers is honored whether it is per partition or total.
      DatasetDescriptor descriptor = view.getDataset().getDescriptor();
      if (descriptor.isPartitioned()) {
        GetStorageKey<E> getKey = new GetStorageKey<E>(view, numPartitionWriters);
        PTable<Pair<GenericData.Record, Integer>, E> table = collection
            .by(getKey, Avros.pairs(Avros.generics(getKey.schema()), Avros.ints()));
        PGroupedTable<Pair<GenericData.Record, Integer>, E> grouped =
            numWriters > 0 ? table.groupByKey(numWriters) : table.groupByKey();
        return grouped.ungroup().values();
      } else {
        return partition(collection, numWriters);
      }
    }
docs: Partitions {@code collection} to be stored efficiently in {@code View}.
  <p>
  This restructures the parallel collection so that all of the entities that
  will be stored in a given partition will be evenly distributed across a specified
  {@code numPartitionWriters}.
  <p>
  If the dataset is not partitioned, then this will structure all of the
  entities to produce a number of files equal to {@code numWriters}.
  
  @param collection a collection of entities
  @param view a {@link View} of a dataset to partition the collection for
  @param numWriters the number of writers that should be used
  @param numPartitionWriters the number of writers data for a single partition will be distributed across
  @param <E> the type of entities in the collection and underlying dataset
  @return an equivalent collection of entities partitioned for the view
  @see #partition(PCollection, View)
  
  @since 1.1.0