Q:

How to generate repetitive groups in pig?

I'm using pig for data preparation, and I have faced a problem which seems to be easy but I can't deal with:
for example, I have a column of names
INPUT:-
 id |  name  
-------------
 1  |  Alicia
 2  |  Ana   
 3  |  Benita
 4  |  Berta 
 5  |  Bertha

And I am expecting Desired OUTPUT:-(can we use FORLOOP Functionality to achieve this?)
  id     |  name  
--------------------------
 1_XX_1  |  Alicia_id_1
 2_XX_1  |  Ana_id_1   
 3_XX_1  |  Benita_id_1
 4_XX_1  |  Berta_id_1
 5_XX_1  |  Bertha_id_1

 1_XX_2  |  Alicia_id_2
 2_XX_2  |  Ana_id_2   
 3_XX_2  |  Benita_id_2
 4_XX_2  |  Berta_id_2
 5_XX_2  |  Bertha_id_2

 1_XX_3  |  Alicia_id_3
 2_XX_3  |  Ana_id_3   
 3_XX_3  |  Benita_id_3
 4_XX_3  |  Berta_id_3
 5_XX_3  |  Bertha_id_3

A:

You could do that with the UDF which would give you some re usability regarding the number of times the input wants to be replicated. Below UDF will do that.
package pigexerciseudf;

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

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;

public class replicateinput extends EvalFunc<DataBag>
{
    public replicateinput()
    {
        
    }
    int rep_factor=0;
    public replicateinput(String a)
    {
        rep_factor=Integer.parseInt(a);
    }
    
    public DataBag exec(Tuple input) throws IOException
    {
        BagFactory bf=BagFactory.getInstance();
        DataBag output=bf.newDefaultBag();
            try
            {
            for(int i=1;i<=rep_factor;i++)
            {
                TupleFactory tp=TupleFactory.getInstance();
                Tuple t1=tp.newTuple(2);
                String key=(String)input.get(0);
                System.out.println("key="+key);
                String value=(String)input.get(1);
                String key_out=key+"_XX_"+i;
                String value_out=value+"_id_"+i;
                t1.set(0,key_out);
                t1.set(1,value_out);
                output.add(t1);
            }
            return output;
            }   
            catch(Exception e)
            {
                throw new IOException(e);
            }
    }

    public Schema outputschema(Schema input)
    {
        try
        {
        List<Schema.FieldSchema> mylist=new ArrayList<Schema.FieldSchema>();
        mylist.add(new Schema.FieldSchema("key_out",DataType.CHARARRAY));
        mylist.add(new Schema.FieldSchema("value_out",DataType.CHARARRAY));
        Schema tupleschema=new Schema(mylist);
        Schema bagschema=new Schema(new Schema.FieldSchema("pair",tupleschema,DataType.TUPLE));
        Schema returnbagsc=new Schema(new Schema.FieldSchema("pairs",bagschema,DataType.BAG));
        return returnbagsc;
        }
        catch(FrontendException e)
        {
            throw new RuntimeException("not able to defime the schema");
        }
    }
}

Input File:

1,Alicia
2,Ana
3,Benita
4,Berta
5,Bertha

REGISTER '/path/to/pigexerciseudf.jar';
define replicat pigexerciseudf.replicateinput('3');                                     
A = LOAD '/home/hduser/exer.dat' using PigStorage(',') as (a:chararray,b:chararray);    
B = FOREACH A GENERATE FLATTEN(replicat(a,b)) as (line:chararray) ;                     
dump B;   

Output:

(1_XX_1,Alicia_id_1)
(1_XX_2,Alicia_id_2)
(1_XX_3,Alicia_id_3)
(2_XX_1,Ana_id_1)
(2_XX_2,Ana_id_2)
(2_XX_3,Ana_id_3)
(3_XX_1,Benita_id_1)
(3_XX_2,Benita_id_2)
(3_XX_3,Benita_id_3)
(4_XX_1,Berta _id_1)
(4_XX_2,Berta _id_2)
(4_XX_3,Berta _id_3)
(5_XX_1,Bertha_id_1)
(5_XX_2,Bertha_id_2)
(5_XX_3,Bertha_id_3)

