从 Hadoop/Cascading 中的 FTP 服务器读取数据答案

【问题标题】：Reading Data From FTP Server in Hadoop/Cascading从 Hadoop/Cascading 中的 FTP 服务器读取数据
【发布时间】：2016-04-20 14:54:26
【问题描述】：

我想从 FTP 服务器读取数据。我以 ftp://Username:Password@host/path 格式提供驻留在 FTP 服务器上的文件的路径。当我使用 map reduce 程序从文件中读取数据时，它工作正常。我想通过级联框架从同一个文件中读取数据。我正在使用级联框架的 Hfs Tap 来读取数据。它抛出以下异常

java.io.IOException: Stream closed
    at org.apache.hadoop.fs.ftp.FTPInputStream.close(FTPInputStream.java:98)
    at java.io.FilterInputStream.close(Unknown Source)
    at org.apache.hadoop.util.LineReader.close(LineReader.java:83)
    at org.apache.hadoop.mapred.LineRecordReader.close(LineRecordReader.java:168)
    at org.apache.hadoop.mapred.MapTask$TrackedRecordReader.close(MapTask.java:254)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:440)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:372)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:212)

下面是我读取文件的级联框架代码：

public class FTPWithHadoopDemo {
    public static void main(String args[]) {
        Tap source = new Hfs(new TextLine(new Fields("line")), "ftp://user:pwd@xx.xx.xx.xx//input1");
        Tap sink = new Hfs(new TextLine(new Fields("line1")), "OP\\op", SinkMode.REPLACE);
        Pipe pipe = new Pipe("First");
        pipe = new Each(pipe, new RegexSplitGenerator("\\s+"));
        pipe = new GroupBy(pipe);
        Pipe tailpipe = new Every(pipe, new Count());
        FlowDef flowDef = FlowDef.flowDef().addSource(pipe, source).addTailSink(tailpipe, sink);
        new HadoopFlowConnector().connect(flowDef).complete();
    }
}

我尝试在 Hadoop 源代码中查找相同的异常。我发现在 MapTask 类中有一个方法 runOldMapper 处理流。并且在相同的方法中，finally 块在流被关闭的位置 (in.close())。当我从 finally 块中删除该行时，它工作正常。下面是代码：

private <INKEY, INVALUE, OUTKEY, OUTVALUE> void runOldMapper(final JobConf job, final TaskSplitIndex splitIndex,
            final TaskUmbilicalProtocol umbilical, TaskReporter reporter)
                    throws IOException, InterruptedException, ClassNotFoundException {
        InputSplit inputSplit = getSplitDetails(new Path(splitIndex.getSplitLocation()), splitIndex.getStartOffset());

        updateJobWithSplit(job, inputSplit);
        reporter.setInputSplit(inputSplit);

        RecordReader<INKEY, INVALUE> in = isSkipping()
                ? new SkippingRecordReader<INKEY, INVALUE>(inputSplit, umbilical, reporter)
                : new TrackedRecordReader<INKEY, INVALUE>(inputSplit, job, reporter);
        job.setBoolean("mapred.skip.on", isSkipping());

        int numReduceTasks = conf.getNumReduceTasks();
        LOG.info("numReduceTasks: " + numReduceTasks);
        MapOutputCollector collector = null;
        if (numReduceTasks > 0) {
            collector = new MapOutputBuffer(umbilical, job, reporter);
        } else {
            collector = new DirectMapOutputCollector(umbilical, job, reporter);
        }
        MapRunnable<INKEY, INVALUE, OUTKEY, OUTVALUE> runner = ReflectionUtils.newInstance(job.getMapRunnerClass(),
                job);

        try {
            runner.run(in, new OldOutputCollector(collector, conf), reporter);
            collector.flush();
        } finally {
            // close
            in.close(); // close input
            collector.close();
        }
    }

请帮我解决这个问题。

谢谢，阿沙达里

【问题讨论】：

标签： java hadoop ftp stream cascading

【解决方案1】：

经过一番努力，我发现 hadoop 使用 org.apache.hadoop.fs.ftp.FTPFileSystem 类进行 FTP。
此类不支持查找，即查找到文件开头的给定偏移量。数据在一个块中读取，然后文件系统寻找下一个块来读取。 FTPFileSystem 的默认块大小为 4KB。由于不支持seek，只能读取小于等于4KB的数据。

【讨论】：