添加並行選項在Pig中失敗作業

Question

我是Pig的新手，試圖在我們的5節點Hadoop集羣上運行以下PigScript。 下面的腳本給我兩列的交集關係中的

register '/home/workspace/Pig/setIntersecUdf.jar'; 
define Inter com.cs.pig.SetIntersection(); 
a = load '/home/pig/pig-0.12.0/input/location.txt' as (location:chararray); 
b = load '/home/pig/pig-0.12.0/input/location.txt' as (location:chararray); 
c = CROSS a,b parallel 10; 
c = DISTINCT c; 
d = Foreach c generate $0,$1,Inter($0,$1) as intersection; 
e = Filter d by intersection !='[]' parallel 10; 
e = Filter e by $0!=$1 parallel 10; 
Store e into '/home/Documents/Pig_output'; 

當我不給「水貨」選項運行此腳本，它運行良好。但是，當我添加此選項時，減速器的tasktracker堆棧中出現以下錯誤，並且作業失敗。

Java.io.IOException: org.apache.hadoop.ipc.RemoteException: 
java.io.IOException: File /tmp/temp-10926921/tmp-1823693600/_temporary/_attempt_201401171541_0001_r_000000_0/part-r-00000 could only be replicated to 0 nodes, instead of 1 at 
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:1639) at 
org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.java:736) at sun.reflect.GeneratedMethodAccessor29.invoke(Unknown Source) at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at 
org.apache.hadoop.ipc.RPC$Server.call(RPC.java:578) at 
org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1393) at 
org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1389) at 
java.security.AccessController.doPrivileged(Native Method) at 
javax.security.auth.Subject.doAs(Subject.java:396) at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1149) at org.apache.hadoop.ipc.Server$Handler.run(Server.java:1387) at 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.runPipeline(PigGenericMapReduce.java:469) at 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.processOnePackageOutput(PigGenericMapReduce.java:432) at 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:404) at 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:256) at 
org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176) at 
org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:650) at 
org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:418) at 
org.apache.hadoop.mapred.Child$4.run(Child.java:255) at 
java.security.AccessController.doPrivileged(Native Method) at 
javax.security.auth.Subject.doAs(Subject.java:396) at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1149) at org.apache.hadoop.mapred.Child.main(Child.java:249) Caused by: 
org.apache.hadoop.ipc.RemoteException: java.io.IOException: File /tmp/temp-10926921/tmp-1823693600/_temporary/_attempt_201401171541_0001_r_000000_0/part-r-00000 could only be replicated to 0 nodes, instead of 1 at 
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:1639) at 
org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.java:736) at 
sun.reflect.GeneratedMethodAccessor29.invoke(Unknown Source) at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at 
org.apache.hadoop.ipc.RPC$Server.call(RPC.java:578) at 
org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1393) at 
org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1389) at 
java.security.AccessController.doPrivileged(Native Method) at 
javax.security.auth.Subject.doAs(Subject.java:396) at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1149) at org.apache.hadoop.ipc.Server$Handler.run(Server.java:1387) at 
org.apache.hadoop.ipc.Client.call(Client.java:1107) at 
org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:229) at 
$Proxy2.addBlock(Unknown Source) at 
sun.reflect.GeneratedMethodAccessor4.invoke(Unknown Source) at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at 
org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:85) at 
org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:62) at $Proxy2.addBlock(Unknown Source) at 
org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.locateFollowingBlock(DFSClient.java:3686) at 
org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.nextBlockOutputStream(DFSClient.java:3546) at 
org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.access$2600(DFSClient.java:2749) at 
org.apache.hadoop.hdfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:2989) 

此外，對於失敗的作業診斷信息當屬：

# of failed Reduce Tasks exceeded allowed limit. FailedCount: 1. LastFailedTask: task_201401171541_0001_r_000002 

請，幫幫忙！

Answer 1

問題似乎是您的減速器上的磁盤空間不足。操作非常昂貴，因爲它寫出了每一對可能的記錄。你有321,372條記錄，所以當你使用CROSS時，你的數據會被炸燬321,372倍。

由於您的數據大約爲6 MB，因此需要編寫大約1.9 TB的數據。當您指定PARALLEL 10時，這會跨越10個縮減器，這大概不會超出任何單個機器的磁盤空間。但是，如果沒有PARALLEL 10，Pig會默認爲單個reducer（可能是因爲您的輸入規模較小），並試圖將所有數據寫入該單個機器。

也許你可以在CROSS之後緊挨着FILTER你的結果，這樣Pig會在嘗試寫入磁盤之前將大部分記錄扔掉。或者您需要重新考慮數據流，以便不需要使用CROSS。