import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
    Configuration configuration = ContextUtil.getConfiguration(jobContext);
    List<InputSplit> splits = new ArrayList<InputSplit>();

    if (isTaskSideMetaData(configuration)) {
        // Although not required by the API, some clients may depend on always
        // receiving ParquetInputSplit. Translation is required at some point.
        for (InputSplit split : super.getSplits(jobContext)) {
            Preconditions.checkArgument(split instanceof FileSplit,
                    "Cannot wrap non-FileSplit: " + split);
            splits.add(ParquetInputSplit.from((FileSplit) split));
        }
        return splits;

    } else {
        splits.addAll(getSplits(configuration, getFooters(jobContext)));
    }

    return splits;
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
public void commitJob(JobContext jobContext) throws IOException {
    super.commitJob(jobContext);
    Configuration conf = ContextUtil.getConfiguration(jobContext);
    Path outputPath = FileOutputFormat.getOutputPath(new JobConf(conf));
    ParquetOutputCommitter.writeMetaDataFile(conf, outputPath);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public static void setUnboundRecordFilter(Job job, Class<? extends UnboundRecordFilter> filterClass) {
    Configuration conf = ContextUtil.getConfiguration(job);
    checkArgument(getFilterPredicate(conf) == null,
            "You cannot provide an UnboundRecordFilter after providing a FilterPredicate");

    conf.set(UNBOUND_RECORD_FILTER, filterClass.getName());
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public RecordReader<Void, T> createRecordReader(
        InputSplit inputSplit,
        TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
    Configuration conf = ContextUtil.getConfiguration(taskAttemptContext);
    ReadSupport<T> readSupport = getReadSupport(conf);
    return new ParquetRecordReader<T>(readSupport, getFilter(conf));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
public ICounter getCounterByNameAndFlag(String groupName, String counterName, String counterFlag) {
    if (ContextUtil.getConfiguration(context).getBoolean(counterFlag, true)) {
        return new MapReduceCounterAdapter(ContextUtil.getCounter(context, groupName, counterName));
    } else {
        return new BenchmarkCounter.NullCounter();
    }
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public void initialize(InputSplit inputSplit, TaskAttemptContext context)
        throws IOException, InterruptedException {
    if (context instanceof TaskInputOutputContext<?, ?, ?, ?>) {
        BenchmarkCounter.initCounterFromContext((TaskInputOutputContext<?, ?, ?, ?>) context);
    } else {
        LOG.error("Can not initialize counter due to context is not a instance of TaskInputOutputContext, but is "
                + context.getClass().getCanonicalName());
    }

    initializeInternalReader(toParquetSplit(inputSplit), ContextUtil.getConfiguration(context));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public void shouldUseParquetFlagToSetCodec(String codecNameStr, CompressionCodecName expectedCodec) throws IOException {

        //Test mapreduce API
        Job job = new Job();
        Configuration conf = job.getConfiguration();
        conf.set(ParquetOutputFormat.COMPRESSION, codecNameStr);
        TaskAttemptContext task = ContextUtil.newTaskAttemptContext(conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
        Assert.assertEquals(CodecConfig.from(task).getCodec(), expectedCodec);

        //Test mapred API
        JobConf jobConf = new JobConf();
        jobConf.set(ParquetOutputFormat.COMPRESSION, codecNameStr);
        Assert.assertEquals(CodecConfig.from(jobConf).getCodec(), expectedCodec);
    }
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public void shouldUseHadoopFlagToSetCodec(String codecClassStr, CompressionCodecName expectedCodec) throws IOException {
    //Test mapreduce API
    Job job = new Job();
    Configuration conf = job.getConfiguration();
    conf.setBoolean("mapred.output.compress", true);
    conf.set("mapred.output.compression.codec", codecClassStr);
    TaskAttemptContext task = ContextUtil.newTaskAttemptContext(conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
    Assert.assertEquals(expectedCodec, CodecConfig.from(task).getCodec());

    //Test mapred API
    JobConf jobConf = new JobConf();
    jobConf.setBoolean("mapred.output.compress", true);
    jobConf.set("mapred.output.compression.codec", codecClassStr);
    Assert.assertEquals(CodecConfig.from(jobConf).getCodec(), expectedCodec);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
protected void setup(Context context) throws IOException, InterruptedException {
  factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ContextUtil.getConfiguration(context)));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public void commitJob(JobContext jobContext) throws IOException {
    super.commitJob(jobContext);
    Configuration configuration = ContextUtil.getConfiguration(jobContext);
    writeMetaDataFile(configuration, outputPath);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public static void setTaskSideMetaData(Job job, boolean taskSideMetadata) {
    ContextUtil.getConfiguration(job).setBoolean(TASK_SIDE_METADATA, taskSideMetadata);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
public static void setReadSupportClass(Job job, Class<?> readSupportClass) {
    ContextUtil.getConfiguration(job).set(READ_SUPPORT_CLASS, readSupportClass.getName());
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
protected List<FileStatus> listStatus(JobContext jobContext) throws IOException {
    return getAllFileRecursively(super.listStatus(jobContext),
            ContextUtil.getConfiguration(jobContext));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * @param jobContext the current job context
 * @return the footers for the files
 * @throws IOException
 */
public List<Footer> getFooters(JobContext jobContext) throws IOException {
    List<FileStatus> statuses = listStatus(jobContext);
    if (statuses.isEmpty()) {
        return Collections.emptyList();
    }
    Configuration config = ContextUtil.getConfiguration(jobContext);
    List<Footer> footers = new ArrayList<Footer>(statuses.size());
    Set<FileStatus> missingStatuses = new HashSet<FileStatus>();
    Map<Path, FileStatusWrapper> missingStatusesMap =
            new HashMap<Path, FileStatusWrapper>(missingStatuses.size());

    if (footersCache == null) {
        footersCache =
                new LruCache<FileStatusWrapper, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
    }
    for (FileStatus status : statuses) {
        FileStatusWrapper statusWrapper = new FileStatusWrapper(status);
        FootersCacheValue cacheEntry =
                footersCache.getCurrentValue(statusWrapper);
        if (Log.DEBUG) {
            LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "")
                    + " found for '" + status.getPath() + "'");
        }
        if (cacheEntry != null) {
            footers.add(cacheEntry.getFooter());
        } else {
            missingStatuses.add(status);
            missingStatusesMap.put(status.getPath(), statusWrapper);
        }
    }
    if (Log.DEBUG) {
        LOG.debug("found " + footers.size() + " footers in cache and adding up "
                + "to " + missingStatuses.size() + " missing footers to the cache");
    }


    if (missingStatuses.isEmpty()) {
        return footers;
    }

    List<Footer> newFooters = getFooters(config, missingStatuses);
    for (Footer newFooter : newFooters) {
        // Use the original file status objects to make sure we store a
        // conservative (older) modification time (i.e. in case the files and
        // footers were modified and it's not clear which version of the footers
        // we have)
        FileStatusWrapper fileStatus = missingStatusesMap.get(newFooter.getFile());
        footersCache.put(fileStatus, new FootersCacheValue(fileStatus, newFooter));
    }

    footers.addAll(newFooters);
    return footers;
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
public Configuration getConfiguration() {
    return ContextUtil.getConfiguration(context);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
@Override
public void increment(long val) {
    ContextUtil.incrementCounter(adaptee, val);
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
protected void setup(org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Void, Group>.Context context) throws java.io.IOException, InterruptedException {
    factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ContextUtil.getConfiguration(context)));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
protected void setup(Context context) throws IOException, InterruptedException {
    factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ContextUtil.getConfiguration(context)));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * set the schema being written to the job conf
 *
 * @param schema        the schema of the data
 * @param configuration the job configuration
 */
public static void setSchema(Job job, MessageType schema) {
    GroupWriteSupport.setSchema(schema, ContextUtil.getConfiguration(job));
}
 

import parquet.hadoop.util.ContextUtil; //导入依赖的package包/类
/**
 * retrieve the schema from the conf
 *
 * @param configuration the job conf
 * @return the schema
 */
public static MessageType getSchema(Job job) {
    return GroupWriteSupport.getSchema(ContextUtil.getConfiguration(job));
}