使用Hadoop 读取大文件，代码怎么写？

以下是使用Hadoop读取大文件的Java代码示例：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.InputFormat;

import java.io.IOException;
import java.io.InputStream;

public class LongInputFormat extends InputFormat<LongWritable, Text> {

    public RecordReader<LongWritable, Text> createRecordReader(
        InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
        throws IOException, InterruptedException {
        return new LongRecordReader();
    }

    @Override
    public List<InputSplit> getSplits(JobContext job) throws IOException, InterruptedException {
        List<InputSplit> splits = new ArrayList<>();
        Configuration conf = job.getConfiguration();
        Path inputDir = getInputPath(job);
        FileSystem fs = inputDir.getFileSystem(conf);
        FileStatus[] status = fs.listStatus(inputDir);

        for (FileStatus file : status) {
            if (file.isFile() && !file.getPath().getName().startsWith(".")) {
                splits.addAll(getSpli tsForFile(file, job.getConfiguration()));
            }
        }

        return splits;
    }

    private static List<FileSplit> getSplitsForFile(FileStatus file, Configuration conf) throws IOException {
        List<FileSplit> splits = new ArrayList<>();
        Path path = file.getPath();
        long length = file.getLen();
        long blockSize = conf.getLong("fs.local.block.size", 67108864);
        long splitSize = conf.getLong("mapred.min.split.size", blockSize);
        long start = 0;
        long bytesRemaining = length;

        while (((double) bytesRemaining) / splitSize > 1.1) {
            splits.add(new FileSplit(path, start, splitSize, (String[]) null));
            start += splitSize;
            bytesRemaining -= splitSize;
        }

        if (bytesRemaining != 0) {
            splits.add(new FileSplit(path, start, bytesRemaining, (String[]) null));
        }

        return splits;
    }

    public static class LongRecordReader extends RecordReader<LongWritable, Text> {
        private LongWritable key = new LongWritable();
        private Text value = new Text();
        private long start;
        private long end;
        private InputStream in;
        private byte[] buffer;
        private String line;
        private int pos;

        public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            FileSplit split = (FileSplit) inputSplit;
            Configuration conf = taskAttemptContext.getConfiguration();
            Path path = split.getPath();
            FileSystem fs = path.getFileSystem(conf);
            this.start = split.getStart();
            this.end = this.start + split.getLength();
            boolean skipFirstLine = false;
            FSDataInputStream fileIn = fs.open(split.getPath());

            if (start != 0) {
                skipFirstLine = true;
                fileIn.seek(start - 1);
            }

            this.in = fileIn;

            if (skipFirstLine) {
                start += in.readLine().length() + 1;
            }

            buffer = new byte[(int) (end - start)];
            IOUtils.readFully(in, buffer, 0, buffer.length);
            line = new String(buffer, "UTF-8");
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            if (pos >= buffer.length) {
                return false;
            }

            int start = pos;

            while (pos < buffer.length && buffer[pos] != '\n') {
                pos++;
            }

            if (pos == buffer.length - 1 && buffer[pos] != '\n') {
                pos++;
            }

            int length = pos - start;

            if (length > 0) {
                line = new String(buffer, start, length, "UTF-8");
                key.set(start);
                value.set(line);
                return true;
            }

            return false;
        }

        @Override
        public LongWritable getCurrentKey() throws IOException, InterruptedException {
            return key;
        }

        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return value;
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            if (end == start) {
                return 0.0f;
            }

            return Math.min(1.0f, (pos - start) / (float) (end - start));
        }

        @Override
        public void close() throws IOException {
            in.close();
        }
    }
}

你可以在你的Mapper类中使用如下代码获取InputSplit、LineOffset和LineContent：

public class DemoMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        String fileName = fileSplit.getPath().getName();
        long offset = fileSplit.getStart() + key.get();
        String line = value.toString().trim();

        // your map code here
    }
}

注意：本代码示例未进行完整测试，仅供参考，请根据实际情况进行修改。