无法使用pig写入sequencefile

mwngjboj 于 2021-06-02 发布在 Hadoop

关注(0)|答案(1)|浏览(366)

我想将一些pig变量存储到hadoop sequencefile中，以便运行外部mapreduce作业。
假设我的数据具有（chararray，int）模式：

(hello,1)
(test,2)
(example,3)

我写了这个存储函数：

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.pig.StoreFunc;
import org.apache.pig.data.Tuple;

public class StoreTest extends StoreFunc {

    private String storeLocation;
    private RecordWriter writer;
    private Job job;

    public StoreTest(){

    }

    @Override
    public OutputFormat getOutputFormat() throws IOException {
        //return new TextOutputFormat();
        return new SequenceFileOutputFormat();
    }

    @Override
    public void setStoreLocation(String location, Job job) throws IOException {
        this.storeLocation = location;
        this.job = job;
        System.out.println("Load location is " + storeLocation);
        FileOutputFormat.setOutputPath(job, new Path(location));        
        System.out.println("Out path " + FileOutputFormat.getOutputPath(job));
    }

    @Override
    public void prepareToWrite(RecordWriter writer) throws IOException {
        this.writer = writer;
    }

    @Override
    public void putNext(Tuple tuple) throws IOException {
        try {
            Text k = new Text(((String)tuple.get(0)));
            IntWritable v = new IntWritable((Integer)tuple.get(1));
            writer.write(k, v);
        } catch (InterruptedException ex) {
            Logger.getLogger(StoreTest.class.getName()).log(Level.SEVERE, null, ex);
        }

    }
}

这只Pig的代码是：

register MyUDFs.jar;
x = load '/user/pinoli/input' as (a:chararray,b:int);
store x into '/user/pinoli/output/' using StoreTest();

但是，存储失败，出现以下错误：

ERROR org.apache.pig.tools.pigstats.PigStats - ERROR 0: java.io.IOException: wrong key class: org.apache.hadoop.io.Text is not class org.apache.hadoop.io.LongWritable

有没有办法解决这个问题？？

hadoop sequencefile apache-pig

来源：https://stackoverflow.com/questions/26611113/fail-to-write-sequencefile-with-pig

1条答案

按热度按时间

ktca8awb1#

问题是您没有设置输出键/值类。你可以在教室里做这个 setStoreLocation() 方法：

@Override
public void setStoreLocation(String location, Job job) throws IOException {
    this.storeLocation = location;
    this.job = job;
    this.job.setOutputKeyClass(Text.class);   // !!!
    this.job.setOutputValueClass(IntWritable.class);  // !!!
    ...

}

我猜您希望将storer与不同的键/值类型一起使用。在这种情况下，可以将它们的类型传递给构造函数。例如：

private Class<? extends WritableComparable> keyClass;
private Class<? extends Writable> valueClass;
...

public StoreTest() {...}

@SuppressWarnings({ "unchecked", "rawtypes" })
    public StoreTest(String keyClass, String valueClass) {
        try {
            this.keyClass = (Class<? extends WritableComparable>) Class.forName(keyClass);
            this.valueClass = (Class<? extends Writable>) Class.forName(valueClass);
        }
        catch (Exception e) {
            throw new RuntimeException("Invalid key/value type", e);
        }
    }

...

 @Override
    public void setStoreLocation(String location, Job job) throws IOException {
        this.storeLocation = location;
        this.job = job;
        this.job.setOutputKeyClass(keyClass);
        this.job.setOutputValueClass(valueClass);
        ...
    }

然后，在pig脚本中设置正确的类型：

register MyUDFs.jar;
DEFINE myStorer StoreTest('org.apache.hadoop.io.Text', 'org.apache.hadoop.io.IntWritable');
x = load '/user/pinoli/input' as (a:chararray,b:int);
store x into '/user/pinoli/output/' using myStorer();

赞(0）回复(0）举报 2021-06-03

我来回答

无法使用pig写入sequencefile

1条答案

相关问题

热门标签

最新问答