You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@crunch.apache.org by "Kiyan Ahmadizadeh (JIRA)" <ji...@apache.org> on 2012/09/21 22:50:07 UTC
[jira] [Created] (CRUNCH-73) Scrunch applications using PipelineApp
do not properly serialize closures to MapReduce tasks.
Kiyan Ahmadizadeh created CRUNCH-73:
---------------------------------------
Summary: Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
Key: CRUNCH-73
URL: https://issues.apache.org/jira/browse/CRUNCH-73
Project: Crunch
Issue Type: Bug
Components: Scrunch
Affects Versions: 0.4.0
Reporter: Kiyan Ahmadizadeh
Assignee: Kiyan Ahmadizadeh
One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
object DivideApp extends PipelineApp {
val divisor = Integer.valueOf(args(0))
val nums = read(From.textFile("numbers.txt"))
val dividedNums = nums.map { n => n / divisor }
dividedNums.write(To.textFile("dividedNums"))
run()
}
Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (CRUNCH-73) Scrunch applications using
PipelineApp do not properly serialize closures to MapReduce tasks.
Posted by "Kiyan Ahmadizadeh (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13460899#comment-13460899 ]
Kiyan Ahmadizadeh commented on CRUNCH-73:
-----------------------------------------
Their analogue to PipelineApp, ScoobiApp, provides a run method for users to override, rather than extending the trait DelayedInit. This avoids DelayedInit's usage of static references.
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
> Attachments: CRUNCH-73-v1.patch, CRUNCH-73-v2.patch
>
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Resolved] (CRUNCH-73) Scrunch applications using
PipelineApp do not properly serialize closures to MapReduce tasks.
Posted by "Josh Wills (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Josh Wills resolved CRUNCH-73.
------------------------------
Resolution: Fixed
Fix Version/s: 0.4.0
Fixed. Thanks Kiyan!
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
> Fix For: 0.4.0
>
> Attachments: CRUNCH-73-v1.patch, CRUNCH-73-v2.patch
>
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (CRUNCH-73) Scrunch applications using PipelineApp
do not properly serialize closures to MapReduce tasks.
Posted by "Kiyan Ahmadizadeh (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Kiyan Ahmadizadeh updated CRUNCH-73:
------------------------------------
Attachment: CRUNCH-73-v2.patch
And here's a patch with a simple test included.
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
> Attachments: CRUNCH-73-v1.patch, CRUNCH-73-v2.patch
>
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (CRUNCH-73) Scrunch applications using
PipelineApp do not properly serialize closures to MapReduce tasks.
Posted by "Kiyan Ahmadizadeh (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13460816#comment-13460816 ]
Kiyan Ahmadizadeh commented on CRUNCH-73:
-----------------------------------------
The problem is that PipelineApp extends the Scala trait DelayedInit, which uses final static references in its implementation.
Scoobi has hit this problem as well. See this note in their user guide: http://nicta.github.com/scoobi/guide/Advanced%20Notes.html#Advanced+Notes
Seeing as the ability to transfer side data in function closures is a big usability win, I propose changing PipelineApp to not use the DelayedInit trait. Instead, clients should override a method
run(args: Array[String]): Unit
to implement their pipeline logic. While this creates a need for a touch more boilerplate code when authoring a PipelineApp, I think the value of sending side data through closures outweighs this.
I've written a patch and verified it works using a pipelineapp I've been developing. I'm attaching the first version for review. I'll attach another version shortly that includes a test.
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (CRUNCH-73) Scrunch applications using
PipelineApp do not properly serialize closures to MapReduce tasks.
Posted by "Josh Wills (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13460895#comment-13460895 ]
Josh Wills commented on CRUNCH-73:
----------------------------------
So the Scoobi rec seems to be "don't use static references." What did they do instead?
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
> Attachments: CRUNCH-73-v1.patch, CRUNCH-73-v2.patch
>
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (CRUNCH-73) Scrunch applications using PipelineApp
do not properly serialize closures to MapReduce tasks.
Posted by "Kiyan Ahmadizadeh (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/CRUNCH-73?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Kiyan Ahmadizadeh updated CRUNCH-73:
------------------------------------
Attachment: CRUNCH-73-v1.patch
First version of fix (test missing but verified in production). Test coming soon.
> Scrunch applications using PipelineApp do not properly serialize closures to MapReduce tasks.
> ---------------------------------------------------------------------------------------------
>
> Key: CRUNCH-73
> URL: https://issues.apache.org/jira/browse/CRUNCH-73
> Project: Crunch
> Issue Type: Bug
> Components: Scrunch
> Affects Versions: 0.4.0
> Reporter: Kiyan Ahmadizadeh
> Assignee: Kiyan Ahmadizadeh
> Attachments: CRUNCH-73-v1.patch
>
>
> One of the great potential advantages of using Scala for writing MapReduce pipelines is the ability to send side data as part of function closures, rather than through Hadoop Configurations or the Distributed Cache. As an absurdly simple example, consider the following Scala PipelineApp that divides all elements of a numeric PCollection by an arbitrary argument:
> object DivideApp extends PipelineApp {
> val divisor = Integer.valueOf(args(0))
> val nums = read(From.textFile("numbers.txt"))
> val dividedNums = nums.map { n => n / divisor }
> dividedNums.write(To.textFile("dividedNums"))
> run()
> }
> Executing this PipelineApp fails. MapReduce tasks get a value of "null" for divisor (or 0 if divisor is forced to be a primitive numeric type). This indicates that an error is occurring in the serialization of Scala function closures that causes unbound variables in the closure to take on their default JVM values.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira