class FacetHandler implements ExtendedDataTypeHandler<KEYIN,KEYOUT,VALUEOUT>, FacetedEstimator<RawRecordContainer> {

    @Override
    public void setup(TaskAttemptContext context) {
        markingFunctions = MarkingFunctions.Factory.createMarkingFunctions();
        
        taskAttemptContext = context;
        
        Configuration conf = context.getConfiguration();
        
        final String t = ConfigurationHelper.isNull(conf, Properties.DATA_NAME, String.class);
        TypeRegistry.getInstance(conf);
        Type type = TypeRegistry.getType(t);
        
        categoryDelimiter = conf.get(type.typeName() + FACET_CATEGORY_DELIMITER, categoryDelimiter);
        
        Map<String,String> categories = conf.getValByRegex(type.typeName() + FACET_CATEGORY_PREFIX_REGEX);
        
        pivotMap = HashMultimap.create();
        
        if (null != categories) {
            for (Map.Entry<String,String> category : categories.entrySet()) {
                final String fields = category.getValue();
                Preconditions.checkNotNull(fields);
                final String[] fieldArray = StringUtils.split(fields, categoryDelimiter.charAt(0));
                Preconditions.checkArgument(fieldArray.length == 2);
                final String pivot = fieldArray[0];
                final String[] facets = StringUtils.split(fieldArray[1], ',');
                pivotMap.putAll(pivot, ImmutableList.copyOf(facets));
            }
        } else {
            throw new IllegalStateException("Categories must be specified");
        }
        
        String predClazzStr = conf.get(FACET_FIELD_PREDICATE_CLASS);
        if (null != predClazzStr) {
            try {
                // Will throw RuntimeException if class can't be coerced into Predicate<String>
                @SuppressWarnings("unchecked")
                Class<Predicate<String>> projClazz = (Class<Predicate<String>>) Class.forName(predClazzStr).asSubclass(Predicate.class);
                fieldFilterPredicate = projClazz.newInstance();
            } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
                throw new RuntimeException(e);
            }
        }
        
        shardIdFactory = new ShardIdFactory(conf);
        facetTableName = new Text(ConfigurationHelper.isNull(conf, FACET_TABLE_NAME, String.class));
        facetMetadataTableName = new Text(conf.get(FACET_METADATA_TABLE_NAME, facetTableName.toString() + "Metadata"));
        facetHashTableName = new Text(conf.get(FACET_HASH_TABLE_NAME, facetTableName.toString() + "Hash"));
        facetHashThreshold = conf.getInt(type.typeName() + FACET_HASH_THRESHOLD, 20);
    }

}