cascalog on hadoop 2.x

[Pavan Reddy]

Hello,
I am currently using cascalog on Hadoop 1.x, and my jobs works fine. I am now trying to use cascalog with Hadoop 2.x, I tested it with one of my jobs (multiple taps, joins etc), they run fine on both Hadoop 1.x and 2.x but is considerably slower (a few times over) on Hadoop 2.x.

I saw another thread here talk about the same issue but there wasn't much discussion about it. Can anyone help me understand how I can fix it? Is there anything I can do to get the jobs running faster on Hadoop 2.x? Thanks.

Pavan

[Pavan Reddy]

I found this thread on the cascading user group [1] and it suggests I should be using cascading-hadoop2-mr1-x.y.z.jar instead of cascading-hadoop-x.y.z.jar since higher versions were used.

And, I see that in the project.clj file of cascalog-core, the dependency is set to [cascading/cascading-hadoop x.y.z] [2]. If I have to use cascading-hadoop2-mr1-x.y.z.jar, should I change this to [cascading/cascading/cascading-hadoop2-mr1 x.y.z]? And should I build cascalog jars by myself after this, to use Hadoop 2.x?
Would that help? please correct me if I am off track here. Thanks.

[1] https://groups.google.com/forum/#!topic/cascading-user/JwxNPJDFbn8
[2] https://github.com/nathanmarz/cascalog/blob/08db2e9afb44278d0b17232e227579fada6e67e7/cascalog-core/project.clj#L24

[Gareth Rogers]

We've also seen this and would be keen to know the answer if you find it.

As AWS are taking away all the 2.x series AMIs and thus everything that installs Hadoop 1 (I believe there's no Hadoop 1 in the 3/4.x series) so this has bumped in priority for us! We're also going to look into upgrading to Cascalog v3 as a solution. Will post back if we get a Cascalog 2 solution working.

Thanks for the pointers!

[Pavan Reddy]

I am working on upgrading cascalog to Hadoop2. I am using -

hadoop-mapreduce-client-core 2.7.1
hadoop-common 2.7.1
cascading-hadoop2-mr1 3.1.0-wip-56

I have made some upgrades based on the changes to the cascading api. But, I am facing an error when I try to run tests.

ERROR in (test-sort) (BaseFlow.java:1012)
Uncaught exception, not in assertion.
expected: nil
  actual: cascading.flow.FlowException: [] unhandled exception
 at cascading.flow.BaseFlow.complete (BaseFlow.java:1012)
    cascalog.cascading.flow$eval9996$fn__9997.invoke (flow.clj:85)
    cascalog.cascading.flow$eval9971$fn__9972$G__9962__9977.invoke (flow.clj:77)
    cascalog.cascading.flow$eval9992$fn__9993.invoke (flow.clj:91)
    cascalog.cascading.flow$eval9971$fn__9972$G__9962__9977.invoke (flow.clj:77)
    cascalog.cascading.flow$eval9988$fn__9989.invoke (flow.clj:95)
    cascalog.cascading.flow$eval9971$fn__9972$G__9962__9977.invoke (flow.clj:77)
    cascalog.cascading.platform$eval10335$fn__10346.invoke (platform.clj:342)
    cascalog.logic.platform$eval7275$fn__7276$G__7262__7285.invoke (platform.clj:8)
    cascalog.api$_QMARK__.doInvoke (api.clj:111)
    clojure.lang.RestFn.applyTo (RestFn.java:137)
    clojure.core$apply.invoke (core.clj:624)
    cascalog.cascading.testing$eval11365$fn__11366.invoke (testing.clj:57)
    cascalog.logic.testing$eval11311$fn__11312$G__11302__11319.invoke (testing.clj:28)
    cascalog.logic.testing$test_QMARK__.doInvoke (testing.clj:33)
    clojure.lang.RestFn.invoke (RestFn.java:421)
    cascalog.api_test/fn (api_test.clj:252)
    clojure.test$test_var$fn__7187.invoke (test.clj:704)
    clojure.test$test_var.invoke (test.clj:704)
    clojure.test$test_vars$fn__7209$fn__7214.invoke (test.clj:722)
    clojure.test$default_fixture.invoke (test.clj:674)
    clojure.test$test_vars$fn__7209.invoke (test.clj:722)
    cascalog.api_test$eval11569$fn__11570.invoke (api_test.clj:14)
    clojure.test$compose_fixtures$fn__7181$fn__7182.invoke (test.clj:681)
    clojure.test$default_fixture.invoke (test.clj:674)
    clojure.test$compose_fixtures$fn__7181.invoke (test.clj:681)
    clojure.test$test_vars.invoke (test.clj:718)
    clojure.test$test_all_vars.invoke (test.clj:728)
    clojure.test$test_ns.invoke (test.clj:747)
    clojure.core$map$fn__4245.invoke (core.clj:2559)
    clojure.lang.LazySeq.sval (LazySeq.java:40)
    clojure.lang.LazySeq.seq (LazySeq.java:49)
    clojure.lang.Cons.next (Cons.java:39)
    clojure.lang.RT.boundedLength (RT.java:1654)
    clojure.lang.RestFn.applyTo (RestFn.java:130)
    clojure.core$apply.invoke (core.clj:626)
    clojure.test$run_tests.doInvoke (test.clj:762)
    clojure.lang.RestFn.applyTo (RestFn.java:137)
    clojure.core$apply.invoke (core.clj:624)
    user$eval1975$fn__2034$fn__2065.invoke (form-init6997287244218067381.clj:1)
    user$eval1975$fn__2034$fn__2035.invoke (form-init6997287244218067381.clj:1)
    user$eval1975$fn__2034.invoke (form-init6997287244218067381.clj:1)
    user$eval1975.invoke (form-init6997287244218067381.clj:1)
    clojure.lang.Compiler.eval (Compiler.java:6703)
    clojure.lang.Compiler.eval (Compiler.java:6693)
    clojure.lang.Compiler.load (Compiler.java:7130)
    clojure.lang.Compiler.loadFile (Compiler.java:7086)
    clojure.main$load_script.invoke (main.clj:274)
    clojure.main$init_opt.invoke (main.clj:279)
    clojure.main$initialize.invoke (main.clj:307)
    clojure.main$null_opt.invoke (main.clj:342)
    clojure.main$main.doInvoke (main.clj:420)
    clojure.lang.RestFn.invoke (RestFn.java:421)
    clojure.lang.Var.invoke (Var.java:383)
    clojure.lang.AFn.applyToHelper (AFn.java:156)
    clojure.lang.Var.applyTo (Var.java:700)
    clojure.main.main (main.java:37)
Caused by: java.io.IOException: Cannot initialize Cluster. Please check your configuration for mapreduce.framework.name and the correspond server addresses.
 at org.apache.hadoop.mapreduce.Cluster.initialize (Cluster.java:120)
    org.apache.hadoop.mapreduce.Cluster.<init> (Cluster.java:82)
    org.apache.hadoop.mapreduce.Cluster.<init> (Cluster.java:75)
    org.apache.hadoop.mapred.JobClient.init (JobClient.java:475)
    org.apache.hadoop.mapred.JobClient.<init> (JobClient.java:454)
    cascading.flow.hadoop.planner.HadoopFlowStepJob.internalNonBlockingStart (HadoopFlowStepJob.java:105)
    cascading.flow.planner.FlowStepJob.blockOnJob (FlowStepJob.java:269)
    cascading.flow.planner.FlowStepJob.start (FlowStepJob.java:184)
    cascading.flow.planner.FlowStepJob.call (FlowStepJob.java:146)
    cascading.flow.planner.FlowStepJob.call (FlowStepJob.java:48)
    java.util.concurrent.FutureTask.run (FutureTask.java:262)
    java.util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor.java:1145)
    java.util.concurrent.ThreadPoolExecutor$Worker.run (ThreadPoolExecutor.java:615)
    java.lang.Thread.run (Thread.java:745)

I figured I need to setup the testing env properly and I tried setting up values of mapreduce.framework.name in job-conf.clj but wasn't able to solve the problem. I tried setting a few more configuration values to set up a local test env based on https://github.com/Cascading/cascading/blob/2.7/cascading-hadoop2-mr1/src/test/java/cascading/platform/hadoop2/Hadoop2MR1Platform.java . But I haven't been ablr to solve the problem.

Any pointers? Thanks.

[Gareth Rogers]

 We've started looking at this and I tried adding into my Clojure Lein project.clj

[cascading/cascading-hadoop2-mr1 "2.5.3"]

[cascalog/cascalog-core "3.0.0" :exclusion [[cascading/cascading-hadoop]]]

I ran two equivalent jobs one with the above and the other with just [cascalog/cascalog-core "3.0.0"] and they both took the same length of time.

Is this similar to what you tried?

I used 2.5.3 as I believe that's what Cascalog is using by default.

Although I've now realised I'm also including cascading.avro/avro-scheme 2.5.0 which is pulling in cascading/cascading-hadoop 2.5.5 having now looked at the output of "lein deps :tree" :( I'll run another test but was curious what people thought of what I've tried so far.

Thanks

Gareth

...

[Pavan Reddy]

Yes, this is similar to what I tried initially, though I hadn't tried excluding [cascading/cascading-hadoop] but it had been considerably slower for me compared to not using hadoop 2.x. Following which I was (and still am, when I find time) trying to make similar changes to cascalog-core project.clj, build my own cascalog-core and use it in my tests.

Let me try using [cascalog/cascalog-core "3.0.0" :exclusion [[cascading/cascading-hadoop]]

] and see how that works for my jobs.

Pavan