1

我有一列是字符串字段。我需要读入这个字符串，将其存储在一个袋子中并给它一个密钥（所以当我将它存储为 JSON 时，它是唯一的。

我的示例数据文件是： “test, kyle”

我希望我的输出看起来像： {"test":[{"key": "test"}, {"value":"kyle"}]}

public class BagTupleExampleUDF extends EvalFunc<DataBag> {

TupleFactory mTupleFactory = TupleFactory.getInstance();
BagFactory mBagFactory = BagFactory.getInstance();
int counter = 0;
static String fieldNames[] = {"key", "value"};


@Override
public DataBag exec(Tuple tuple) throws IOException {
    // expect one string
    if (tuple == null || tuple.size() == 0) {
        throw new IllegalArgumentException("BagTupleExampleUDF: requires input parameters.");
    }
    try {

        String key = (String) tuple.get(0);
        String input = (String) tuple.get(1);

        DataBag output = mBagFactory.newDefaultBag();

            output.add(mTupleFactory.newTuple(Collections.singletonList(key)));
            output.add(mTupleFactory.newTuple(Collections.singletonList(input)));


        return output;
    }
    catch (Exception e) {
        throw new IOException("BagTupleExampleUDF: caught exception processing input.", e);
    }
}

public Schema outputSchema(Schema input) {
    // Function returns a bag with this schema: { (Double), (Double) }
    // Thus the outputSchema type should be a Bag containing a Double
    try{

        Schema bagSchema = new Schema();
        String schemaName =  getSchemaName(this.getClass().getName().toLowerCase(), input);

        bagSchema.add(new Schema.FieldSchema(fieldNames[counter], DataType.CHARARRAY));
        counter++;
        return new Schema(new Schema.FieldSchema(schemaName, bagSchema, DataType.BAG));
        }
    catch (Exception e){
        throw new RuntimeException(e);
    }
}

我遇到的问题是这一行：

bagSchema.add(new Schema.FieldSchema(fieldNames[counter], DataType.CHARARRAY));

基本上，我从元组中读取的每个值都需要一个不同的识别键，这样当使用 Pig 完成时，我可以引用我添加的这些新列。

我对 Pig 尤其是 UDFS 还是很陌生，所以如果需要更多信息，请告诉我。

4

0 回答 0