在pig中使用UDF从数据库中读取数据

Question

我需要从数据库读取数据，并使用PIG分析数据。我用java编写了一个UDF，引用了下面的链接

register /tmp/UDFJars/CassandraUDF_1-0.0.1-SNAPSHOT-jar-with-dependencies.jar;
A = Load '/user/sampleFile.txt' using udf.DBLoader('10.xx.xxx.4','username','password','select * from customer limit 10') as (f1 : chararray);
DUMP A;


package udf;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.pig.LoadFunc;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import com.data.ConnectionCassandra;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;

public class DBLoader extends LoadFunc {
    private final Log log = LogFactory.getLog(getClass());
    Session session;
    private ArrayList mProtoTuple = null;
    private String jdbcURL;
    private String user;
    private String pass;
    private int count = 0;
    private String query;
    ResultSet result;
    List<Row> rows;
    int colSize;
    protected TupleFactory mTupleFactory = TupleFactory.getInstance();

    public DBLoader() {
    }

    public DBLoader(String jdbcURL, String user, String pass, String query) {

        this.jdbcURL = jdbcURL;
        this.user = user;
        this.pass = pass;
        this.query = query;

    }

    @Override
    public InputFormat getInputFormat() throws IOException {
        log.info("Inside InputFormat");
        // TODO Auto-generated method stub
        try {
            return new TextInputFormat();
        } catch (Exception exception) {
            log.error(exception.getMessage());
            log.error(exception.fillInStackTrace());
            throw new IOException();
        }
    }

    @Override
    public Tuple getNext() throws IOException {
        log.info("Inside get Next");
        Row row = rows.get(count);
        if (row != null) {
            mProtoTuple = new ArrayList<Object>();
            for (int colNum = 0; colNum < colSize; colNum++) {
                mProtoTuple.add(row.getObject(colNum));
            }
        } else {
            return null;
        }
        Tuple t = mTupleFactory.newTuple(mProtoTuple);
        mProtoTuple.clear();
        return t;

    }

    @Override
    public void prepareToRead(RecordReader arg0, PigSplit arg1) throws IOException {
        log.info("Inside Prepare to Read");
        session = null;
        if (query == null) {
            throw new IOException("SQL Insert command not specified");
        }
        if (user == null || pass == null) {
            log.info("Creating Session with user name and password as: " + user + " : " + pass);
            session = ConnectionCassandra.connectToCassandra1(jdbcURL, user, pass);
            log.info("Session Created");
        } else {
            session = ConnectionCassandra.connectToCassandra1(jdbcURL, user, pass);
        }
        log.info("Executing Query " + query);
        result = session.execute(query);
        log.info("Query Executed :" + query);
        rows = result.all();
        count = 0;
        colSize = result.getColumnDefinitions().asList().size();
    }

    @Override
    public void setLocation(String location, Job job) throws IOException {
        log.info("Inside Set Location");
        try {
            FileInputFormat.setInputPaths(job, location);
        } catch (Exception exception) {
            log.info("Some thing went wrong : " + exception.getMessage());
            log.debug(exception);
        }

    }
}

org.apache.pig.impl.logicallayer.frontendException：错误1066：无法在org.apache.pig.pig.tools.grunt.gruntparser.openiterator（pigserver.java:892)在org.apache.pig.tools.gruntparser.processdump（gruntparser.java:774)在org.apache.pig.tools.pigscript.parse（pigscriptparser.java:372)在.parseStoponError（gruntParser.java:173)在org.apache.pig.tools.grunt.grunt.exec（grunt.java:84)在org.apache.pig.main.run（main.java:484)在org.apache.pig.main.run（main.java:484)在org.apache.pig.main.main（main.java:158)在sun.reflect.nativeMethodAccessorImpl.Invoke（原生方法）在606）在org.apache.hadoop.util.runjar.run（runjar.java:221)在org.apache.hadoop.util.runjar.main（runjar.java：136）原因：html" target="_blank">java.io.ioException：在org.apache.pig.pigserver.openiterator（pigserver.java：884）上，以异常状态终止的作业失败

Answer 1

维韦克！你有准备读吗？（我看到您做了一些日志记录，所以知道日志中实际有什么会很好）此外，提供完整的stacktrace也是非常好的，因为我看到您没有完整的底层异常。只是一些想法--我从来没有尝试过在编写LoadFunc时不实现自己的InputFormat和recordReader-textInputFormat检查文件的存在和大小（并根据文件大小创建许多InputSplits)，所以如果虚拟文件为空，很有可能不会产生InputSplits，或者产生零长度的InputSplits。因为它的长度为零，所以可能会导致pig抛出异常。因此，最好的建议是实现自己的InputFormat（实际上很简单）。也只是一个快速的尝试-尝试

set pig.splitCombination false

可能不会有帮助，但很容易尝试。