Pig 中的 SimpleTextLoader UDF

Posted 2023-04-14

【中文标题】Pig 中的 SimpleTextLoader UDF

【英文标题】：SimpleTextLoader UDF in Pig

【发布时间】：2014-11-25 09:58:51

【问题描述】：

我想为 Pig UDF 创建一个自定义加载函数，我使用链接创建了一个 SimpleTextLoader

https://pig.apache.org/docs/r0.11.0/udf.html ，我已经成功生成了这段代码的 jar 文件，在 pig 中注册并运行 Pig 脚本。我得到的是空输出。我不知道如何解决这个问题，任何帮助将不胜感激。

下面是我的 Java 代码

public class SimpleTextLoader extends LoadFunc
    protected RecordReader in = null;
    private byte fieldDel = '\t';
    private ArrayList<Object> mProtoTuple = null;
    private TupleFactory mTupleFactory = TupleFactory.getInstance();
    private static final int BUFFER_SIZE = 1024;
      public SimpleTextLoader() 
    

public SimpleTextLoader(String delimiter) 

        this();
        if (delimiter.length() == 1) 
            this.fieldDel = (byte)delimiter.charAt(0);
         else if (delimiter.length() > 1 && delimiter.charAt(0) == '\\') 
            switch (delimiter.charAt(1)) 
            case 't':
                this.fieldDel = (byte)'\t';
                break;

            case 'x':
               fieldDel =
                    Integer.valueOf(delimiter.substring(2), 16).byteValue();
               break;

            case 'u':
                this.fieldDel =
                    Integer.valueOf(delimiter.substring(2)).byteValue();
                break;

            default:
                throw new RuntimeException("Unknown delimiter " + delimiter);
            
         else 
            throw new RuntimeException("PigStorage delimeter must be a single character");
        
    

private void readField(byte[] buf, int start, int end) 
        if (mProtoTuple == null) 
            mProtoTuple = new ArrayList<Object>();
        

        if (start == end) 
            // NULL value
            mProtoTuple.add(null);
         else 
            mProtoTuple.add(new DataByteArray(buf, start, end));
        
        @Override
    public Tuple getNext() throws IOException 
         try 
            boolean notDone = in.nextKeyValue();
            if (notDone) 
                return null;
            
            Text value = (Text) in.getCurrentValue();
            System.out.println("printing value"  +value);
            byte[] buf = value.getBytes();
            int len = value.getLength();
            int start = 0;

            for (int i = 0; i < len; i++) 
                if (buf[i] == fieldDel) 
                    readField(buf, start, i);
                    start = i + 1;
                
            
            // pick up the last field
            readField(buf, start, len);

            Tuple t =  mTupleFactory.newTupleNoCopy(mProtoTuple);
            mProtoTuple = null;
            System.out.println(t);
            return t;
         catch (InterruptedException e) 
            int errCode = 6018;
            String errMsg = "Error while reading input";
            throw new ExecException(errMsg, errCode,
                    PigException.REMOTE_ENVIRONMENT, e);
        

    


    @Override
    public void setLocation(String string, Job job) throws IOException 
        FileInputFormat.setInputPaths(job,string);
    

    @Override
    public InputFormat getInputFormat() throws IOException 
      return new TextInputFormat();
    

    @Override
    public void prepareToRead(RecordReader reader, PigSplit ps) throws IOException 
        in=reader;
    

下面是我的猪脚本

REGISTER /home/hadoop/netbeans/sampleloader/dist/sampleloader.jar
a= load '/input.txt' using sampleloader.SimpleTextLoader();
store a into 'output';

【问题讨论】：

您是否收到任何错误日志？

没有得到任何错误日志，作业执行成功但创建了空文件。

【参考方案1】：

您正在使用 sampleloader.SimpleTextLoader()，它什么都不做，因为它只是一个空构造函数。 而是使用sampleloader.SimpleTextLoader(String delimiter)，它正在执行拆分的实际操作。

【讨论】：

您好，感谢您的回复，我也尝试过使用分隔符，但输出为空。

使用猪自定义加载函数Here

，请参考此链接逐步加载数据