我远离我现在使用的集群,所以我不能 100% 确定,但这应该是正确的路径:
-- Assuming myData.txt is formatted like:
-- 1 A b
-- 2 B c
-- etc.
A = LOAD 'myData.txt' USING PigStorage(' ')
AS (number: int, val1: chararray, val2: chararray) ;
STORE A INTO 'myOutputDir'
-- Stores using \t as the input separator
USING org.apache.pig.piggybank.storage.MultiStorage('myOutputDir', '0') ;
如果这样做,则将创建 3 个目录(用于 1、2 和 3),并且在这些目录中,只有与文件夹名称相同编号的文件位于它们之下。但是,在这些目录中的每一个中都可以有许多不同的文件(每个映射器/减速器一个)。此外,还必须存储字段 0。因此,输出可能如下所示:
--myOutputDir
|
|-->1
| |-->1-00000 #Contains 1 A a
| |-->1-00001 #Contains 1 A b
|
|-->2
| |-->2-00000 #Contains 2 B b
| |-->2-00001 #Contains 2 B c
|
|-->3
| |-->3-00000 #Contains 3 A a, 3 B b
| |-->3-00001 #Contains 3 C c
|
3-00000的内容:
3 A a
3 B b
但是,因为您知道输出文件的名称,所以您可以加载您创建的每个输出目录并根据需要格式化它们:
-- Repeat this for all the numbers
A3 = LOAD 'myOutputDir/3' AS (number: int, val1: chararray, val2: chararray) ;
B3 = FOREACH A3 GENERATE val1, val2 ;
STORE B3 INTO 'myOutputDir/stripped3' ;
所以现在输出将如下所示:
A a
B b
C c
但是根据映射器作业的数量,数据仍然可以在多个文件中拆分。如果它们需要全部在同一个文件中,我只建议编写一个将这些部分合并在一起的脚本。我使用这样的东西(但显然更通用):
import os
import glob
partfiles = os.path.join('myOutputDir', 'stripped3', 'part-m-[0-9]*')
with open('part-m-COMPLETE-3', 'w') as outfile:
for myfile in glob.glob(partfiles):
with open(myfile, 'r') as infile:
for line in infile:
outfile.write(line)