过滤customercode=455的输入，生成所需的2列，然后按customerid分组，然后使用bagtostring。

B = FILTER A BY Customercode == 455;
C = FOREACH B GENERATE $0 as CustomerId,$4 as CustomerText;
D = GROUP C BY CustomerId;
E = FOREACH D GENERATE group AS CustomerId, BagToString(C.CustomerText, '|'); -- Note:This will generate 1001,SPRINT56|Unlimited Plan|2017-05-05 so,you will have to concat the first field with '|' and then concat the resulting field with the second field which is already delimited by '|'.
F = FOREACH E GENERATE CONCAT(CONCAT($0,'|'),$1);
DUMP F;

rawData = LOAD data;
filteredData = FILTER rawData BY (Customercode == 455);

--Extract and set Val/Description/Date based on CustomerText and 'null' otherwise
ExtractedData = FOREACH filteredData GENERATE
            customerId,
            (CustomerType == "CODE" ? CustomerText : null) AS Val,
            (CustomerType == "DESC" ? CustomerText : null) AS Description,
            (CustomerType == "DATE" ? CustomerText : null) AS Date;

groupedData = GROUP ExtractedData  BY customerId;

--While taking MAX, all 'nulls' will be ignored
finalData = FOREACH groupedData GENERATE
             group as CustomerId,
             MAX($1.Val) AS Val,
             MAX($1.Description) AS Description,
             MAX($1.Date) AS Date;

DUMP finalData;

我已经详细说明了核心逻辑。加载、格式化和存储应该是直截了当的。