public CreateBloomFilter(Pipe keys, String bloomFilterID, String approxCountPartsDir, String bloomPartsDir, String keyBytesField) throws IOException {
// Collect stats used to configure the bloom filter creation step
Pipe smallPipe = new Each(keys, new CollectKeyStats(keyBytesField));
smallPipe = new Each(smallPipe, new Fields(keyBytesField), new GetIndices(HashFunctionFactory.DEFAULT_HASH_FACTORY), new Fields("split", "index", "hash_num"));
smallPipe = new Each(smallPipe, new Fields("split", "index", "hash_num"), new Unique.FilterPartialDuplicates());
smallPipe = new GroupBy(smallPipe, new Fields("split"));
smallPipe = new Every(smallPipe, new Fields("index", "hash_num"), new CreateBloomFilterFromIndices(), Fields.ALL);
ConfigDef bloomDef = smallPipe.getStepConfigDef();