DataStax Support Forums » Topic: Submitting a Map-Reduce Job using CassandraJobConf

DataStax Support Forums » Topic: Submitting a Map-Reduce Job using CassandraJobConf http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf Software, Support, and Training for Apache Cassandra en-US Sat, 18 May 2013 23:59:52 +0000 http://bbpress.org/?v=1.0.3 <![CDATA[Search]]> q http://www.datastax.com/support-forums/search.php deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-272 Wed, 06 Jul 2011 14:55:01 +0000 deltafoxtrot 272@http://www.datastax.com/support-forums/ Thanks for your feedback. Let me take a step back for a second and explain what we had in place prior to our Brisk integration effort. Our product is comprised of three web apps (each of which is the product of multiple projects - all Maven-based). Two of the three web apps handle the OLTP side of our product and one of the web apps handles the OLAP side. About a month ago, we replaced one of our OLAP tasks with a Hadoop implementation and saw significant performance improvement. This Hadoop task lived within the web app and was executed via the ToolRunner as the result of a restful request to the web app. All of this worked albeit with limitations. So given the promise of Brisk, we're quite keen to integrate. Unfortunately we did not find any examples on your site that relate to the way we submitted jobs previously. When you write our service "seems very non-standard", how do you mean specifically? What is the "standard" way to submit jobs? Command line? I've aligned the versions of Hadoop to use 0.20.203-brisk1. As we're a Maven shop, we pushed the ivy/hadoop-core.pom.xml that we found in your beta 1 distribution when creating an artifact for the brisk version of Hadoop. Not sure if this is an accurate pom for the brisk hadoop core code. We also added the core-site.xml to our web app (the same one that came down with the Brisk beta 1 binary). It initially failed (ClassNotFound for SnappyException) though we added that dependency to overcome. Then it complained that it couldn't find the cassandra.yaml. Needless to say, it seems this is not the recommended approach. :o) tjake on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-271 Wed, 06 Jul 2011 12:21:51 +0000 tjake 271@http://www.datastax.com/support-forums/ I the version of hadoop you are using on your service the same as brisk (you should be using the brisk jars) Also, you need to use the core-site.xml in your config. I don't see cfs:// anywhere in your config log statements. I don't quite understand what you are doing in terms of your service but it seems very non-standard :) deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-270 Wed, 06 Jul 2011 10:53:38 +0000 deltafoxtrot 270@http://www.datastax.com/support-forums/ The current run method with a temporarily hard-coded JobTracker address is posted below. It fails downstream from the Job instantiation. @Override public int run(String[] args) throws Exception { try { Configuration conf = new Configuration(); conf.set("mapred.job.tracker", "ondroid-si-1:8012"); new GenericOptionsParser(conf, args).getRemainingArgs(); ConfigHelper.setRpcPort(conf, Integer.valueOf(port).toString()); ConfigHelper.setInitialAddress(conf, host); ConfigHelper.setPartitioner(conf, RandomPartitioner.class.getCanonicalName()); ConfigHelper.setInputColumnFamily(conf, IN_KEYSPACE, IN_COLUMN_FAMILY); ConfigHelper.setOutputColumnFamily(conf, OUT_KEYSPACE, OUT_COLUMN_FAMILY); SlicePredicate predicate = new SlicePredicate(); predicate.setSlice_range(new SliceRange(bytes(null), bytes(null), false, Integer.MAX_VALUE)); ConfigHelper.setInputSlicePredicate(conf, predicate); Job job = new Job(conf, "dependency.graph"); job.setJarByClass(DependencyGraph.class); job.setMapperClass(DependencyGraphMapper.class); job.setReducerClass(DependencyGraphReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(ByteBuffer.class); job.setOutputValueClass(List.class); job.setOutputFormatClass(ColumnFamilyOutputFormat.class); job.setInputFormatClass(ColumnFamilyInputFormat.class); if (log.isInfoEnabled()) { log.info("Submitting DependencyGraph job..."); } job.waitForCompletion(true); return 0; } catch (Exception e) { log.error("Caught exception waiting for job to complete:\n" + ExceptionUtils.getStackTrace(e)); throw e; } } deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-269 Wed, 06 Jul 2011 10:37:02 +0000 deltafoxtrot 269@http://www.datastax.com/support-forums/ Ok. We've set the JobTracker address based on the response from brisktool. When we execute now it fails with an EOFException (see detailed logging and stacktrace below). Any ideas? From our application log: 06-07-2011 11:31:40 [http-8080-1] INFO controller.DataController - Computing targets graphs started at: Wed Jul 06 11:31:40 BST 2011 06-07-2011 11:31:40 [http-8080-1] INFO cassandra.AbstractStore - Truncating column family 'data_graphs' 06-07-2011 11:31:40 [http-8080-1] INFO client.CassandraClient - CassandraProxyClient(localhost, 9170, true, ROUND_ROBIN) 06-07-2011 11:31:40 [http-8080-1] INFO client.CassandraClient - Connected to cassandra at localhost:9170 06-07-2011 11:31:40 [http-8080-1] INFO hadoop.DependencyGraph - Configuring new DependencyGraph job against host localhost and port 9170 06-07-2011 11:31:40 [http-8080-1] INFO hadoop.DependencyGraph - Configuration Properties: io.seqfile.compress.blocksize=1000000, fs.checkpoint.size=67108864, io.skip.checksum.errors=false, mapred.used.genericoptionsparser=true, fs.s3n.impl=org.apache.hadoop.fs.s3native.NativeS3FileSystem, fs.s3.maxRetries=4, webinterface.private.actions=false, fs.s3.impl=org.apache.hadoop.fs.s3.S3FileSystem, hadoop.native.lib=true, fs.checkpoint.edits.dir=${fs.checkpoint.dir}, ipc.server.listen.queue.size=128, fs.default.name=file:///, ipc.client.idlethreshold=4000, fs.hsftp.impl=org.apache.hadoop.hdfs.HsftpFileSystem, hadoop.tmp.dir=/tmp/hadoop-${user.name}, fs.checkpoint.dir=${hadoop.tmp.dir}/dfs/namesecondary, fs.s3.block.size=67108864, hadoop.security.authorization=false, io.serializations=org.apache.hadoop.io.serializer.WritableSerialization, hadoop.util.hash.type=murmur, io.seqfile.lazydecompress=true, io.file.buffer.size=4096, io.mapfile.bloom.size=1048576, fs.s3.buffer.dir=${hadoop.tmp.dir}/s3, hadoop.logfile.size=10000000, ipc.client.kill.max=10, io.compression.codecs=org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.BZip2Codec, topology.script.number.args=100, fs.har.impl=org.apache.hadoop.fs.HarFileSystem, io.seqfile.sorter.recordlimit=1000000, fs.trash.interval=0, local.cache.size=10737418240, ipc.server.tcpnodelay=false, ipc.client.connect.max.retries=10, fs.ramfs.impl=org.apache.hadoop.fs.InMemoryFileSystem, hadoop.rpc.socket.factory.class.default=org.apache.hadoop.net.StandardSocketFactory, fs.kfs.impl=org.apache.hadoop.fs.kfs.KosmosFileSystem, fs.checkpoint.period=3600, topology.node.switch.mapping.impl=org.apache.hadoop.net.ScriptBasedMapping, hadoop.logfile.count=10, fs.ftp.impl=org.apache.hadoop.fs.ftp.FTPFileSystem, fs.file.impl=org.apache.hadoop.fs.LocalFileSystem, fs.hdfs.impl=org.apache.hadoop.hdfs.DistributedFileSystem, ipc.client.connection.maxidletime=10000, io.mapfile.bloom.error.rate=0.005, io.bytes.per.checksum=512, mapred.job.tracker=ondroid-si-1:8012, fs.har.impl.disable.cache=true, ipc.client.tcpnodelay=false, fs.hftp.impl=org.apache.hadoop.hdfs.HftpFileSystem, fs.s3.sleepTimeSeconds=10, 06-07-2011 11:31:41 [http-8080-1] ERROR hadoop.DependencyGraph - Caught exception waiting for job to complete: java.io.IOException: Call to ondroid-si-1/10.20.5.191:8012 failed on local exception: java.io.EOFException at org.apache.hadoop.ipc.Client.wrapException(Client.java:775) at org.apache.hadoop.ipc.Client.call(Client.java:743) at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220) at org.apache.hadoop.mapred.$Proxy105.getProtocolVersion(Unknown Source) at org.apache.hadoop.ipc.RPC.getProxy(RPC.java:359) at org.apache.hadoop.mapred.JobClient.createRPCProxy(JobClient.java:429) at org.apache.hadoop.mapred.JobClient.init(JobClient.java:423) at org.apache.hadoop.mapred.JobClient.<init>(JobClient.java:410) at org.apache.hadoop.mapreduce.Job.<init>(Job.java:50) at org.apache.hadoop.mapreduce.Job.<init>(Job.java:54) at ntoklo.matrix.impl.computation.hadoop.DependencyGraph.run(DependencyGraph.java:87) at ntoklo.matrix.impl.controller.DataController.computeGraphs(DataController.java:206) at ntoklo.matrix.impl.MatrixImpl.serviceComputeGraphs(MatrixImpl.java:89) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.lang.reflect.Method.invoke(Unknown Source) at com.sun.jersey.server.impl.model.method.dispatch.AbstractResourceMethodDispatchProvider$TypeOutInvoker._dispatch(AbstractResourceMethodDispatchProvider.java:167) at com.sun.jersey.server.impl.model.method.dispatch.ResourceJavaMethodDispatcher.dispatch(ResourceJavaMethodDispatcher.java:70) at com.sun.jersey.server.impl.uri.rules.HttpMethodRule.accept(HttpMethodRule.java:279) at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:136) at com.sun.jersey.server.impl.uri.rules.ResourceClassRule.accept(ResourceClassRule.java:86) at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:136) at com.sun.jersey.server.impl.uri.rules.RootResourceClassesRule.accept(RootResourceClassesRule.java:74) at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1347) at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1279) at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1229) at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1219) at com.sun.jersey.spi.container.servlet.WebComponent.service(WebComponent.java:419) at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:537) at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:699) at javax.servlet.http.HttpServlet.service(HttpServlet.java:717) at com.google.inject.servlet.ServletDefinition.doService(ServletDefinition.java:216) at com.google.inject.servlet.ServletDefinition.service(ServletDefinition.java:141) at com.google.inject.servlet.ManagedServletPipeline.service(ManagedServletPipeline.java:93) at com.google.inject.servlet.FilterChainInvocation.doFilter(FilterChainInvocation.java:63) at com.google.inject.servlet.ManagedFilterPipeline.dispatch(ManagedFilterPipeline.java:122) at com.google.inject.servlet.GuiceFilter.doFilter(GuiceFilter.java:110) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206) at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:233) at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:191) at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:127) at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:102) at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:109) at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:298) at org.apache.coyote.http11.Http11Processor.process(Http11Processor.java:859) at org.apache.coyote.http11.Http11Protocol$Http11ConnectionHandler.process(Http11Protocol.java:588) at org.apache.tomcat.util.net.JIoEndpoint$Worker.run(JIoEndpoint.java:489) at java.lang.Thread.run(Unknown Source) Caused by: java.io.EOFException at java.io.DataInputStream.readInt(Unknown Source) at org.apache.hadoop.ipc.Client$Connection.receiveResponse(Client.java:501) at org.apache.hadoop.ipc.Client$Connection.run(Client.java:446) From the Cassandra/Brisk system.log: WARN [pool-3-thread-1] 2011-07-06 11:34:10,110 Server.java (line 1110) Incorrect header or version mismatch from 10.20.5.191:57394 got version 3 expected version 4 tjake on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-266 Tue, 05 Jul 2011 16:59:34 +0000 tjake 266@http://www.datastax.com/support-forums/ No, but this explains the problem :) You need to know ahead of time what the jobtracker address is. Assuming you have run: brisktool jobtracker and have the location you must set this as the value of the property mapred.job.tracker in your configuration... deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-265 Tue, 05 Jul 2011 16:54:13 +0000 deltafoxtrot 265@http://www.datastax.com/support-forums/ It is executed via a call to a restful endpoint in a web app. We just create a new instance of the class that extends Tool and execute the run method. Are we required to use the command line to schedule MR jobs in Brisk? tjake on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-263 Tue, 05 Jul 2011 16:31:09 +0000 tjake 263@http://www.datastax.com/support-forums/ How are you executing the hadoop job? You should be using "brisk hadoop jar myjar.jar ..." deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-262 Tue, 05 Jul 2011 16:28:57 +0000 deltafoxtrot 262@http://www.datastax.com/support-forums/ Thanks again for your response. The updated run method is now: @Override public int run(String[] args) throws Exception { Configuration conf = new Configuration(); new GenericOptionsParser(conf, args).getRemainingArgs(); // conf.set("mapred.job.tracker", CassandraJobConf.getJobTrackerNode().getHostName() + ":8012"); // conf.set("mapreduce.jobtracker.address", CassandraJobConf.getJobTrackerNode().getHostName() + ":8012"); Job job = new Job(conf, "dependency.graph"); [...] However, it doesn't appear that Brisk sets the JobTracker host. I've dumped the configuration properties before the job is submitted. 05-07-2011 17:21:20 [http-8080-1] INFO controller.DataController - Computing targets graphs started at: Tue Jul 05 17:21:20 BST 2011 05-07-2011 17:21:20 [http-8080-1] INFO cassandra.AbstractStore - Truncating column family 'data_graphs' 05-07-2011 17:21:20 [http-8080-1] INFO hadoop.DependencyGraph - Configuring new DependencyGraph job against host localhost and port 9170 05-07-2011 17:21:20 [http-8080-1] INFO hadoop.DependencyGraph - Configuration Properties: io.seqfile.compress.blocksize=1000000 fs.checkpoint.size=67108864 io.skip.checksum.errors=false mapred.used.genericoptionsparser=true fs.s3n.impl=org.apache.hadoop.fs.s3native.NativeS3FileSystem fs.s3.maxRetries=4 webinterface.private.actions=false fs.s3.impl=org.apache.hadoop.fs.s3.S3FileSystem hadoop.native.lib=true fs.checkpoint.edits.dir=${fs.checkpoint.dir} ipc.server.listen.queue.size=128 fs.default.name=file:/// ipc.client.idlethreshold=4000 fs.hsftp.impl=org.apache.hadoop.hdfs.HsftpFileSystem hadoop.tmp.dir=/tmp/hadoop-${user.name} fs.checkpoint.dir=${hadoop.tmp.dir}/dfs/namesecondary fs.s3.block.size=67108864 hadoop.security.authorization=false io.serializations=org.apache.hadoop.io.serializer.WritableSerialization hadoop.util.hash.type=murmur io.seqfile.lazydecompress=true io.file.buffer.size=4096 io.mapfile.bloom.size=1048576 fs.s3.buffer.dir=${hadoop.tmp.dir}/s3 hadoop.logfile.size=10000000 ipc.client.kill.max=10 io.compression.codecs=org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.BZip2Codec topology.script.number.args=100 fs.har.impl=org.apache.hadoop.fs.HarFileSystem io.seqfile.sorter.recordlimit=1000000 fs.trash.interval=0 local.cache.size=10737418240 ipc.server.tcpnodelay=false ipc.client.connect.max.retries=10 fs.ramfs.impl=org.apache.hadoop.fs.InMemoryFileSystem hadoop.rpc.socket.factory.class.default=org.apache.hadoop.net.StandardSocketFactory fs.kfs.impl=org.apache.hadoop.fs.kfs.KosmosFileSystem fs.checkpoint.period=3600 topology.node.switch.mapping.impl=org.apache.hadoop.net.ScriptBasedMapping hadoop.logfile.count=10 fs.ftp.impl=org.apache.hadoop.fs.ftp.FTPFileSystem fs.file.impl=org.apache.hadoop.fs.LocalFileSystem fs.hdfs.impl=org.apache.hadoop.hdfs.DistributedFileSystem ipc.client.connection.maxidletime=10000 io.mapfile.bloom.error.rate=0.005 io.bytes.per.checksum=512 fs.har.impl.disable.cache=true ipc.client.tcpnodelay=false fs.hftp.impl=org.apache.hadoop.hdfs.HftpFileSystem fs.s3.sleepTimeSeconds=10 If I manually set the mapreduce.jobtracker.address property on the configuration (e.g., conf.set("mapreduce.jobtracker.address", CassandraJobConf.getJobTrackerNode().getHostName() + ":8012")), then an exception will be thrown to indicate it can't find the cassandra.yaml (see below). 05-07-2011 17:07:29 [http-8080-1] ERROR config.DatabaseDescriptor - Fatal configuration error org.apache.cassandra.config.ConfigurationException: Cannot locate cassandra.yaml at org.apache.cassandra.config.DatabaseDescriptor.getStorageConfigURL(DatabaseDescriptor.java:111) at org.apache.cassandra.config.DatabaseDescriptor.<clinit>(DatabaseDescriptor.java:121) at org.apache.cassandra.db.ColumnFamily.getComparatorFor(ColumnFamily.java:397) at org.apache.cassandra.db.ReadCommand.getComparator(ReadCommand.java:94) at org.apache.cassandra.db.SliceByNamesReadCommand.<init>(SliceByNamesReadCommand.java:44) at org.apache.cassandra.db.SliceByNamesReadCommand.<init>(SliceByNamesReadCommand.java:38) at org.apache.cassandra.hadoop.trackers.TrackerManager.getCurrentJobtrackerLocation(TrackerManager.java:51) at org.apache.cassandra.hadoop.trackers.CassandraJobConf.getJobTrackerNode(CassandraJobConf.java:62) at ntoklo.matrix.impl.computation.hadoop.DependencyGraph.run(DependencyGraph.java:77) at ntoklo.matrix.impl.controller.DataController.computeGraphs(DataController.java:206) at ntoklo.matrix.impl.MatrixImpl.serviceComputeGraphs(MatrixImpl.java:89) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.lang.reflect.Method.invoke(Unknown Source) at com.sun.jersey.server.impl.model.method.dispatch.AbstractResourceMethodDispatchProvider$TypeOutInvoker._dispatch(AbstractResourceMethodDispatchProvider.java:167) at com.sun.jersey.server.impl.model.method.dispatch.ResourceJavaMethodDispatcher.dispatch(ResourceJavaMethodDispatcher.java:70) at com.sun.jersey.server.impl.uri.rules.HttpMethodRule.accept(HttpMethodRule.java:279) at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:136) at com.sun.jersey.server.impl.uri.rules.ResourceClassRule.accept(ResourceClassRule.java:86) at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:136) at com.sun.jersey.server.impl.uri.rules.RootResourceClassesRule.accept(RootResourceClassesRule.java:74) at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1347) at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1279) at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1229) at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1219) at com.sun.jersey.spi.container.servlet.WebComponent.service(WebComponent.java:419) at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:537) at com.sun.jersey.spi.container.servlet.ServletContainer.service(ServletContainer.java:699) at javax.servlet.http.HttpServlet.service(HttpServlet.java:717) If I make the brisk cassandra.yaml available to the web app classpath, then I get the following error: 05-07-2011 17:12:43 [http-8080-1] INFO config.DatabaseDescriptor - Loading settings from file:/opt/SP/apps/ntoklo/services%23matrix/WEB-INF/classes/cassandra.yaml 05-07-2011 17:12:43 [http-8080-1] INFO config.DatabaseDescriptor - DiskAccessMode 'auto' determined to be mmap, indexAccessMode is mmap 05-07-2011 17:12:43 [http-8080-1] ERROR hadoop.DependencyGraph - java.lang.IllegalArgumentException: Unknown ColumnFamily jobtracker in keyspace brisk_system Any thoughts? And again, many thanks for your assistance! tjake on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-261 Tue, 05 Jul 2011 14:57:19 +0000 tjake 261@http://www.datastax.com/support-forums/ Ah looks like you need to add the following to your main class (where conf is the JobConf instance): String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); This will allow brisk to set the jobtracker host. deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-260 Tue, 05 Jul 2011 14:47:15 +0000 deltafoxtrot 260@http://www.datastax.com/support-forums/ Thanks for responding so quickly! After swapping out the CassandraJobConf for the deprecated JobConf, we still cannot see the job in the JobTracker or TaskTracker sites. The log output is posted below. We only have a single node configured at the moment in our test environment. We can start the node with or without Hadoop enabled and the job still runs as seen below. Anything else you might suggest? 05-07-2011 15:38:31 [http-8080-1] INFO controller.DataController - Computing targets graphs started at: Tue Jul 05 15:38:31 BST 2011 05-07-2011 15:38:31 [http-8080-1] INFO cassandra.AbstractStore - Truncating column family 'data_graphs' 05-07-2011 15:38:31 [http-8080-1] INFO hadoop.DependencyGraph - Configuring new DependencyGraph job against host localhost and port 9170 05-07-2011 15:38:32 [http-8080-1] INFO jvm.JvmMetrics - Initializing JVM Metrics with processName=JobTracker, sessionId= 05-07-2011 15:38:32 [http-8080-1] INFO hadoop.DependencyGraph - Submitting DependencyGraph job... 05-07-2011 15:38:32 [http-8080-1] WARN mapred.JobClient - Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 05-07-2011 15:38:33 [http-8080-1] INFO mapred.JobClient - Running job: job_local_0001 05-07-2011 15:38:33 [Thread-20] INFO mapred.MapTask - io.sort.mb = 100 05-07-2011 15:38:33 [Thread-20] INFO mapred.MapTask - data buffer = 79691776/99614720 05-07-2011 15:38:33 [Thread-20] INFO mapred.MapTask - record buffer = 262144/327680 05-07-2011 15:38:34 [http-8080-1] INFO mapred.JobClient - map 0% reduce 0% 05-07-2011 15:38:37 [Thread-20] INFO mapred.MapTask - Spilling map output: record full = true 05-07-2011 15:38:37 [Thread-20] INFO mapred.MapTask - bufstart = 0; bufend = 5613510; bufvoid = 99614720 05-07-2011 15:38:37 [Thread-20] INFO mapred.MapTask - kvstart = 0; kvend = 262144; length = 327680 05-07-2011 15:38:38 [SpillThread] INFO mapred.MapTask - Finished spill 0 [...] 05-07-2011 15:42:51 [Thread-20] INFO mapred.LocalJobRunner - reduce > reduce 05-07-2011 15:42:51 [Thread-20] INFO mapred.TaskRunner - Task 'attempt_local_0001_r_000000_0' done. 05-07-2011 15:42:51 [http-8080-1] INFO mapred.JobClient - map 100% reduce 100% 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Job complete: job_local_0001 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Counters: 12 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - FileSystemCounters 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - FILE_BYTES_READ=6703653590 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - FILE_BYTES_WRITTEN=12441116494 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Map-Reduce Framework 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Reduce input groups=555796 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Combine output records=0 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Map input records=1728478 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Reduce shuffle bytes=0 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Reduce output records=555796 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Spilled Records=56052510 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Map output bytes=317681794 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Combine input records=0 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Map output records=14839085 05-07-2011 15:42:52 [http-8080-1] INFO mapred.JobClient - Reduce input records=14839085 tjake on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-259 Tue, 05 Jul 2011 12:47:20 +0000 tjake 259@http://www.datastax.com/support-forums/ You don't need to pass the CassandraJobConf, just use the regular JobConf. Implementing Tool is the only required step. deltafoxtrot on "Submitting a Map-Reduce Job using CassandraJobConf" http://www.datastax.com/support-forums/topic/submitting-a-map-reduce-job-using-cassandrajobconf#post-258 Tue, 05 Jul 2011 12:15:30 +0000 deltafoxtrot 258@http://www.datastax.com/support-forums/ Apologies in advance if this seems a bit 101, but I'm struggling to submit a Map-Reduce job to the Brisk-managed JobTracker via code. The class that represents the MR job extends Configured and implements Tool. In its run method, I pass a new CassandraJobConf to the Job constructor (see code below). While the job runs, it seems to spawn it's own Hadoop framework to process the job as opposed to passing the job along to the Brisk-managed JobTracker (it also doesn't show in the JobTracker or TaskTracker web apps). I haven't seen an example using the CassandraJobConf so not sure if this is the intended usage. Any thoughts would be greatly appreciated. @Override public int run(String[] args) throws Exception { Job job = new Job(new CassandraJobConf(), "dependency.graph"); job.setJarByClass(DependencyGraph.class); job.setMapperClass(DependencyGraphMapper.class); job.setReducerClass(DependencyGraphReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(ByteBuffer.class); job.setOutputValueClass(List.class); job.setOutputFormatClass(ColumnFamilyOutputFormat.class); job.setInputFormatClass(ColumnFamilyInputFormat.class); ConfigHelper.setRpcPort(job.getConfiguration(), Integer.valueOf(port).toString()); ConfigHelper.setInitialAddress(job.getConfiguration(), host); ConfigHelper.setPartitioner(job.getConfiguration(), RandomPartitioner.class.getCanonicalName()); ConfigHelper.setInputColumnFamily(job.getConfiguration(), IN_KEYSPACE, IN_COLUMN_FAMILY); ConfigHelper.setOutputColumnFamily(job.getConfiguration(), OUT_KEYSPACE, OUT_COLUMN_FAMILY); SlicePredicate predicate = new SlicePredicate(); predicate.setSlice_range(new SliceRange(bytes(null), bytes(null), false, Integer.MAX_VALUE)); ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate); job.waitForCompletion(true); return 0; }