pig-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "liyunzhang_intel (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (PIG-5054) Initialize SchemaTupleBackend correctly in backend in spark mode if spark job has more than 1 stage
Date Tue, 15 Nov 2016 19:17:58 GMT

    [ https://issues.apache.org/jira/browse/PIG-5054?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15667989#comment-15667989
] 

liyunzhang_intel commented on PIG-5054:
---------------------------------------

[~szita]:  yes,this issue can only be reproduced in yarn mode. And i reproduced this on a
cluster with 3 nodes.

> Initialize SchemaTupleBackend  correctly in backend in spark mode if spark job has more
than 1 stage
> ----------------------------------------------------------------------------------------------------
>
>                 Key: PIG-5054
>                 URL: https://issues.apache.org/jira/browse/PIG-5054
>             Project: Pig
>          Issue Type: Sub-task
>          Components: spark
>            Reporter: liyunzhang_intel
>            Assignee: Adam Szita
>             Fix For: spark-branch
>
>
> After PIG-4970, we remove the serialization and deserialization of jobConf in spark mode.
But in script of pigmix L5.pig 
> {code}
>   register pigperf.jar;
> A = load '/user/pig/tests/data/pigmix/page_views' using org.apache.pig.test.udf.storefunc.PigPerformanceLoader()
>     as (user, action, timespent, query_term, ip_addr, timestamp,
>         estimated_revenue, page_info, page_links);
> B = foreach A generate user;
> alpha = load '/user/pig/tests/data/pigmix/users' using PigStorage('\u0001') as (name,
phone, address,
>         city, state, zip);
> beta = foreach alpha generate name;
> C = cogroup beta by name, B by user parallel 40;
> D = filter C by COUNT(beta) == 0;
> E = foreach D generate group;
> store E into 'L5out';
> {code}
> following error is thrown out in log
> {noformat}
> java.lang.RuntimeException: org.apache.pig.backend.executionengine.ExecException: ERROR
0: Error while executing ForEach at [C[-1,-1]]
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.OutputConsumerIterator.readNext(OutputConsumerIterator.java:89)
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.OutputConsumerIterator.hasNext(OutputConsumerIterator.java:96)
>      at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:41)
>      at scala.collection.Iterator$$anon$14.hasNext(Iterator.scala:388)
>      at scala.collection.convert.Wrappers$IteratorWrapper.hasNext(Wrappers.scala:29)
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.OutputConsumerIterator.readNext(OutputConsumerIterator.java:57)
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.OutputConsumerIterator.hasNext(OutputConsumerIterator.java:96)
>      at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:41)
>      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply$mcV$sp(PairRDDFunctions.scala:1111)
>      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1111)
>      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1111)
>      at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1250)
>      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1119)
>      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1091)
>      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
>      at org.apache.spark.scheduler.Task.run(Task.scala:89)
>      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
>      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
>      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
>      at java.lang.Thread.run(Thread.java:722)
> Caused by: org.apache.pig.backend.executionengine.ExecException: ERROR 0: Error while
executing ForEach at [C[-1,-1]]
>      at org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.getNextTuple(POForEach.java:329)
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.ForEachConverter$ForEachFunction$1$1.getNextResult(ForEachConverter.java:87)
>      at org.apache.pig.backend.hadoop.executionengine.spark.converter.OutputConsumerIterator.readNext(OutputConsumerIterator.java:69)
>      ... 20 more
> Caused by: java.lang.RuntimeException: initialize was not called! Even when SchemaTuple
feature is not set, it should be called.
>      at org.apache.pig.data.SchemaTupleBackend.newSchemaTupleFactory(SchemaTupleBackend.java:294)
>      at org.apache.pig.data.SchemaTupleFactory.getInstance(SchemaTupleFactory.java:119)
>      at org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.processPlan(POForEach.java:350)
>      at org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.getNextTuple(POForEach.java:321)
>      ... 22 more
> {noformat}
> It seems that SchemaTupleBackend is not correctly initialized.  The reason for this error
is because after PIG-4970, we initialized SchemaTupleBackend in [PigInputFormatSpark#initialize|https://github.com/apache/pig/blob/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/running/PigInputFormatSpark.java#L68]
before we load data(stage-0). But it is not initialized in other stage(such as stage1). So
if there are more than 1 stage, the exception will be thrown out.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message