A = foreach X generate .. as field1:chararray, .. as field2:bag{};
A = load '..' using PigStorage('\t', '-schema');
A = load '..' using org.apache.pig.piggybank.storage.avro.AvroStorage();

private static final TupleFactory TUPLE_FACTORY = TupleFactory.getInstance();
Tuple t = TUPLE_FACTORY.newTuple(s);

// factory
BagFactory mBagFactory = BagFactory.getInstance();
DataBag output = mBagFactory.newDefaultBag();

// if you know upfront how many tuples you are going to put in this bag.
DataBag bag = new NonSpillableDataBag(m.size());

public DefaultDataBag() {
    mContents = new ArrayList<Tuple>();
}

public SortedDataBag(Comparator<Tuple> comp) {
mComp = (comp == null) ? new DefaultComparator() : comp;
mContents = new ArrayList<Tuple>();
}

public DistinctDataBag() {
mContents = new HashSet<Tuple>();
}

'dvc': 'imei_123',
'appUse': {
    'app name1': {
        ...
        'timelist': [...]
    },
    'app name2': {
        ...
        'timelist': [...]
    },
    ...
}£¬
'appInstall': {
    'app name1': {
        ...
        'timelist': [...]
    },
    ...
}

public class AppTimelist  extends EvalFunc<DataBag>{
    private static final TupleFactory TUPLE_FACTORY = TupleFactory.getInstance();
    private static final BagFactory BAG_FACTORY = BagFactory.getInstance();
    
    @SuppressWarnings({ "unchecked" })
    @Override
    public DataBag exec(Tuple input) throws IOException {
        Map<String, Map<String, Object>> m = (Map<String, Map<String, Object>>) input.get(0);
        List<Object> result = new ArrayList<Object>();
        DataBag output = BAG_FACTORY.newDefaultBag();
                    
        if(m == null) 
            return null;
        
        for(Map.Entry<String, Map<String, Object>> e: m.entrySet()) {
            result.clear();
            String app = e.getKey();
            long size = ((DataBag) e.getValue().get("timelist")).size();
            result.add(app);
            result.add(size);
            output.add(TUPLE_FACTORY.newTuple(result));
        }
        
        return output;
    }
}

public class DistinctBag extends EvalFunc<DataBag> {
    BagFactory mBagFactory = BagFactory.getInstance();
    
    @Override
    public DataBag exec(Tuple input) throws IOException {
        if(input == null || input.size() == 0) {
            return null;
        }
        
        DataBag in = (DataBag) input.get(0);
        DataBag out = mBagFactory.newDistinctBag();
        
        if(in == null) {
            return null;
        }
        
        for(Tuple tp: in) {
            DataBag applist = (DataBag) tp.get(0);
            for(Tuple app: applist) 
                out.add(app);
        }   
        return out;
    }
}

// AppTimelist.java
@Override
public Schema outputSchema(Schema input) {
    try {
        Schema tupleSchema = new Schema();
        FieldSchema chararrayFieldSchema = new Schema.FieldSchema(null, DataType.CHARARRAY);
        FieldSchema longFieldSchema = new Schema.FieldSchema(null, DataType.LONG);
        tupleSchema.add(chararrayFieldSchema);
        tupleSchema.add(longFieldSchema);
        return new Schema(new Schema.FieldSchema(getSchemaName(this
                .getClass().getName().toLowerCase(), input), tupleSchema,
                DataType.TUPLE));
    } catch (Exception e) {
        return null;
    }
}

// DistinctBag.java
@Override
public Schema outputSchema(Schema input) {
FieldSchema innerFieldSchema = new Schema.FieldSchema(null, DataType.CHARARRAY);
Schema innerSchema = new Schema(innerFieldSchema);
Schema bagSchema = null;

try {
bagSchema = new Schema(new FieldSchema(null, innerSchema, DataType.BAG));
} catch(FrontendException e) {
throw new RuntimeException(e);
}
return bagSchema;
}

define AvroStorage org.apache.pig.piggybank.storage.avro.AvroStorage;
define DistinctBag com.pig.udf.bag.DistinctBag;
A = load '..' using AvroStorage();
B = foreach A generate value.fields.data#'dvc' as dvc:chararray, value.fields.data#'appInstall' as ins:map[map[]];
C = foreach B generate dvc, KEYSET(ins) as applist;
D = group C by dvc;
-- extract applist from grouped D
E = foreach D {
    projected = foreach $1 generate applist;
    generate group as dvc, projected as grouped;
}
F = foreach E generate dvc, DistinctBag(grouped) as applist;
store F into '..' using AvroStorage();

define AvroStorage org.apache.pig.piggybank.storage.avro.AvroStorage;
define AppTimelist com.pig.udf.map.AppTimelist;
A = load '..' using AvroStorage();
B = foreach A generate value.fields.data#'dvc' as dvc:chararray, value.fields.data#'appUse' as use:map[map[]];
C = foreach B generate dvc, flatten(AppTimelist(use)) as (app, fre);
D = group C by (dvc, app);
E = foreach D generate flatten(group) as (dvc, app), SUM($1.fre) as fre;
F = group E by dvc;
G = foreach F {
        projected = foreach $1 generate app, fre;
        generate group as dvc, projected as appfre;
}
store G into '..' using AvroStorage();