1

私のhadoop 2.2.0プログラムでPDFファイルを解析したいのですが、これを見つけて、それが言うことに従いました。今まで、次の3つのクラスがあります。

  1. PDFWordCount: map 関数と reduce 関数を含むメイン クラス。(ネイティブの Hadoop ワードカウントサンプルと同様ですが、代わりにクラスTextInputFormatを使用しました。PDFInputFormat

  2. PDFRecordReader extends RecordReader<LongWritable, Text>：ここでの主な作業はどれですか。特にinitialize、より多くの説明のためにここに関数を配置します。

    public void initialize(InputSplit genericSplit, TaskAttemptContext context)
      throws IOException, InterruptedException {
      System.out.println("initialize");
      System.out.println(genericSplit.toString());
    FileSplit split = (FileSplit) genericSplit;
    System.out.println("filesplit convertion has been done");
    final Path file = split.getPath();
    Configuration conf = context.getConfiguration();
    conf.getInt("mapred.linerecordreader.maxlength", Integer.MAX_VALUE);
    FileSystem fs = file.getFileSystem(conf);
    System.out.println("fs has been opened");
    start = split.getStart();
    end = start + split.getLength();
    System.out.println("going to open split");
    FSDataInputStream filein = fs.open(split.getPath());
    System.out.println("going to load pdf");
    PDDocument pd = PDDocument.load(filein);
    System.out.println("pdf has been loaded");
    PDFTextStripper stripper = new PDFTextStripper();
    in =
    new LineReader(new ByteArrayInputStream(stripper.getText(pd).getBytes(
        "UTF-8")));
    start = 0;
    this.pos = start;
    System.out.println("init has finished");
  }

    system.out.println(デバッグ用に s を見ることができます。このメソッドは への変換genericSplitに失敗しますFileSplit。コンソールに最後に表示されるのは、次のとおりです。

    hdfs://localhost:9000/in:0+9396432

    これはgenericSplit.toString()

  3. PDFInputFormat extends FileInputFormat<LongWritable, Text>:メソッドで作成するだけ new PDFRecordReaderです。createRecordReader

私の間違いを知りたいですか？

追加のクラスか何かが必要ですか？

4

2 に答える 2

1

PDF の読み取りはそれほど難しくありません。クラス FileInputFormat と RecordReader を拡張する必要があります。FileInputClass はバイナリであるため、PDF ファイルを分割できないはずです。

public class PDFInputFormat extends FileInputFormat<Text, Text> {

  @Override
  public RecordReader<Text, Text> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException, InterruptedException {
      return new PDFLineRecordReader();
  }

  // Do not allow to ever split PDF files, even if larger than HDFS block size
  @Override
  protected boolean isSplitable(JobContext context, Path filename) {
    return false;
  }

}

次に、RecordReader が読み取り自体を実行します (私は PDFBox を使用して PDF を読み取ります)。

public class PDFLineRecordReader extends RecordReader<Text, Text> {

private Text key = new Text();
private Text value = new Text();
private int currentLine = 0;
private List<String> lines = null;

private PDDocument doc = null;
private PDFTextStripper textStripper = null;

@Override
public void initialize(InputSplit split, TaskAttemptContext context)
        throws IOException, InterruptedException {

    FileSplit fileSplit = (FileSplit) split;
    final Path file = fileSplit.getPath();

    Configuration conf = context.getConfiguration();
    FileSystem fs = file.getFileSystem(conf);
    FSDataInputStream filein = fs.open(fileSplit.getPath());

    if (filein != null) {

        doc = PDDocument.load(filein);

        // Konnte das PDF gelesen werden?
        if (doc != null) {
            textStripper = new PDFTextStripper();
            String text = textStripper.getText(doc);

            lines = Arrays.asList(text.split(System.lineSeparator()));
            currentLine = 0;

        }

    }
}

    // False ends the reading process
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {

    if (key == null) {
        key = new Text();
    }

    if (value == null) {
        value = new Text();
    }

    if (currentLine < lines.size()) {
        String line = lines.get(currentLine);

        key.set(line);

        value.set("");
        currentLine++;

        return true;
    } else {

        // All lines are read? -> end
        key = null;
        value = null;
        return false;
    }
}

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

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

@Override
public float getProgress() throws IOException, InterruptedException {
    return (100.0f / lines.size() * currentLine) / 100.0f;
}

@Override
public void close() throws IOException {

    // If done close the doc
    if (doc != null) {
        doc.close();
    }

}

お役に立てれば！

于 2014-02-07T11:32:57.090 に答える
0 
package com.sidd.hadoop.practice.pdf;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.sidd.hadoop.practice.input.pdf.PdfFileInputFormat;
import com.sidd.hadoop.practice.output.pdf.PdfFileOutputFormat;

public class ReadPdfFile {

    public static class MyMapper extends
            Mapper<LongWritable, Text, LongWritable, Text> {
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
//          context.progress();
            context.write(key, value);
        }
    }

    public static class MyReducer extends
            Reducer<LongWritable, Text, LongWritable, Text> {
        public void reduce(LongWritable key, Iterable<Text> values,
                           Context context) throws IOException, InterruptedException {
            if (values.iterator().hasNext()) {
                context.write(key, values.iterator().next());
            } else {
                context.write(key, new Text(""));
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        Job job = new Job(conf, "Read Pdf");
        job.setJarByClass(ReadPdfFile.class);

        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(PdfFileInputFormat.class);
        job.setOutputFormatClass(PdfFileOutputFormat.class);

        removeDir(args[1], conf);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static void removeDir(String path, Configuration conf) throws IOException {
        Path output_path = new Path(path);
        FileSystem fs = FileSystem.get(conf);

        if (fs.exists(output_path)) {
            fs.delete(output_path, true);
        }
    }   

}
于 2016-12-23T09:33:08.323 に答える