import org.apache.crunch.MapFn; //导入依赖的package包/类
public <T> PCollection<Pair<Integer, T>> apply(PCollection<T> pcollect) {
  PTypeFamily ptf = pcollect.getTypeFamily();
  PType<Pair<Integer, T>> pt = ptf.pairs(ptf.ints(), pcollect.getPType());
  return pcollect.parallelDo("crossfold", new MapFn<T, Pair<Integer, T>>() {
    private transient RandomGenerator rand;
    
    @Override
    public void initialize() {
      if (rand == null) {
        this.rand = RandomManager.getSeededRandom(seed);
      }
    }
    
    @Override
    public Pair<Integer, T> map(T t) {
      return Pair.of(rand.nextInt(numFolds), t);
    }
    
  }, pt);
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
public static <K, T, N extends Number> PTable<K, T> groupedWeightedSample(
    PTable<K, Pair<T, N>> input,
    int sampleSize,
    RandomGenerator random) {
  PTypeFamily ptf = input.getTypeFamily();
  PType<K> keyType = input.getPTableType().getKeyType();
  @SuppressWarnings("unchecked")
  PType<T> ttype = (PType<T>) input.getPTableType().getValueType().getSubTypes().get(0);
  PTableType<K, Pair<Double, T>> ptt = ptf.tableOf(keyType, ptf.pairs(ptf.doubles(), ttype));

  // fill reservoirs by mapping over the vectors and re-emiting them; each map task emits at most sampleSize
  // vectors per fold; the combiner/reducer will combine the outputs and pare down to sampleSize vectors total
  PTable<K, Pair<Double, T>> samples = input.parallelDo("reservoirSampling",
      new SampleFn<K, T, N>(sampleSize, random, ttype), ptt);

  // pare down to just a single reservoir with sampleSize vectors
  PTable<K, Pair<Double, T>> reservoir = samples.groupByKey(1).combineValues(new WRSCombineFn<K, T>(sampleSize, ttype));

  // strip the weights off the final sampled reservoir and return
  return reservoir.parallelDo("strippingSamplingWeights", new MapFn<Pair<K, Pair<Double, T>>, Pair<K, T>>() {
    @Override
    public Pair<K, T> map(Pair<K, Pair<Double, T>> p) {
      return Pair.of(p.first(), p.second().second());
    }
  }, ptf.tableOf(keyType, ttype));
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
/**
 * Swap the key and value part of a PTable. The original PTypes are used in the opposite order
 * @param table PTable to process
 * @param <K> Key type (will become value type)
 * @param <V> Value type (will become key type)
 * @return PType&lt;V, K&gt; containing the same data as the original
 */
public static <K, V> PTable<V, K> swapKeyValue(PTable<K, V> table) {
  PTypeFamily ptf = table.getTypeFamily();
  return table.parallelDo(new MapFn<Pair<K, V>, Pair<V, K>>() {
    @Override
    public Pair<V, K> map(Pair<K, V> input) {
      return Pair.of(input.second(), input.first());
    }
  }, ptf.tableOf(table.getValueType(), table.getKeyType()));
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
/**
 * When creating toplists, it is often required to sort by count descending. As some sort operations don't support
 * order (such as SecondarySort), this method will negate counts so that a natural-ordered sort will produce a
 * descending order.
 * @param table PTable to process
 * @param <K> key type
 * @return PTable of the same format with the value negated
 */
public static <K> PTable<K, Long> negateCounts(PTable<K, Long> table) {
  return table.parallelDo(new MapFn<Pair<K, Long>, Pair<K, Long>>() {
    @Override
    public Pair<K, Long> map(Pair<K, Long> input) {
      return Pair.of(input.first(), -input.second());
    }
  }, table.getPTableType());
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
@Test
public void testZScores() {
  PCollection<Record> elems = VECS.parallelDo(new MapFn<RealVector, Record>() {
    @Override
    public Record map(RealVector vec) {
      return new VectorRecord(vec);
    }
  }, null);
  Summarizer sr = new Summarizer();
  Summary s = sr.build(elems).getValue();
  StandardizeFn fn = new StandardizeFn(s, Transform.Z);
  assertEquals(ImmutableList.of(Vectors.of(-1, 1),
      Vectors.of(-1, -1), Vectors.of(1, -1),
      Vectors.of(1, 1)), elems.parallelDo(fn, MLAvros.vector()).materialize());
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
@Test
public void testMissing() throws Exception {
  PCollection<Record> elems = STRINGS.parallelDo(new MapFn<String, Record>() {
    @Override
    public Record map(String input) {
      return new CSVRecord(Arrays.asList(input.split(",")));
    }
  }, MLRecords.csvRecord(AvroTypeFamily.getInstance(), ","));
  Summarizer sr = new Summarizer();
  Summary s = sr.build(elems).getValue();
  assertEquals(1, s.getStats(1).getMissing());
  assertEquals(2.0, s.getStats(1).mean(), 0.01);
  assertEquals(0.0, s.getStats(1).stdDev(), 0.01);
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
@Test
public void testTrailingIgnoredFields() throws Exception {
  Spec spec = RecordSpec.builder().add("field1", DataType.DOUBLE)
      .add("field2", DataType.DOUBLE).add("field3", DataType.DOUBLE).build();
  PCollection<Record> elems = STRINGS.parallelDo(new MapFn<String, Record>() {
    @Override
    public Record map(String input) {
      return new CSVRecord(Arrays.asList(input.split(",")));
    }
  }, MLRecords.csvRecord(AvroTypeFamily.getInstance(), ","));
  Summarizer sr = new Summarizer().spec(spec).ignoreColumns(2);
  sr.build(elems).getValue();
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
public static AvroType<Record> record(Schema schema) {
  return Avros.derived(Record.class,
      new MapFn<GenericData.Record, Record>() {
        @Override
        public Record map(GenericData.Record gdr) {
          GenericData.Record copy = new GenericData.Record(gdr, true);
          return new AvroRecord(copy);
        }
      },
      new AvroRecordFn(schema),
      Avros.generics(schema));
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
public static PType<Record> vectorRecord(PType<RealVector> ptype, boolean sparse) {
  return ptype.getFamily().derived(Record.class,
      new MapFn<RealVector, Record>() {
        @Override
        public Record map(RealVector v) {
          return new VectorRecord(v);
        }
      },
      new Record2VectorFn(sparse),
      ptype);
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
public static <T> PCollection<T> sample(
    PCollection<T> input,
    int sampleSize,
    RandomGenerator random) {
  PTypeFamily ptf = input.getTypeFamily();
  PType<Pair<T, Integer>> ptype = ptf.pairs(input.getPType(), ptf.ints());
  return weightedSample(
      input.parallelDo(new MapFn<T, Pair<T, Integer>>() {
        @Override
        public Pair<T, Integer> map(T t) { return Pair.of(t, 1); }
      }, ptype),
      sampleSize,
      random);
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
public static <T, N extends Number> PCollection<T> weightedSample(
    PCollection<Pair<T, N>> input,
    int sampleSize,
    RandomGenerator random) {
  PTypeFamily ptf = input.getTypeFamily();
  PTable<Integer, Pair<T, N>> groupedIn = input.parallelDo(
      new MapFn<Pair<T, N>, Pair<Integer, Pair<T, N>>>() {
        @Override
        public Pair<Integer, Pair<T, N>> map(Pair<T, N> p) {
          return Pair.of(0, p);
        }
      }, ptf.tableOf(ptf.ints(), input.getPType()));
  return groupedWeightedSample(groupedIn, sampleSize, random).values();
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
private static MapFn<CQLRecord, ByteBuffer> makeKeyFn(final int[] partitionKeyIndexes) {
  return new MapFn<CQLRecord, ByteBuffer>() {
    @Override
    public ByteBuffer map(final CQLRecord record) {
      return CassandraRecordUtils.getPartitionKey(record.getValues(), partitionKeyIndexes);
    }
  };
}
 

import org.apache.crunch.MapFn; //导入依赖的package包/类
/**
 * @return a map function to extract the partition key from a record
 */
public MapFn<CQLRecord, ByteBuffer> getKeyFn() {
  return makeKeyFn(clusterInfo.getPartitionKeyIndexes());
}