Support Forums : DataStax

DataStax Support Forums » DataStax Enterprise

Submitting a Map-Reduce Job using CassandraJobConf

(12 posts) (2 voices)

Started 1 year ago by deltafoxtrot
Latest reply from deltafoxtrot
This topic is not resolved

Tags:

deltafoxtrot
Member

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;
}

Posted 1 year ago #
tjake
Moderator

You don't need to pass the CassandraJobConf, just use the regular JobConf.

Implementing Tool is the only required step.

Posted 1 year ago #
deltafoxtrot
Member

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

Posted 1 year ago #
tjake
Moderator

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.

Posted 1 year ago #
deltafoxtrot
Member

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!

Posted 1 year ago #
tjake
Moderator

How are you executing the hadoop job?

You should be using "brisk hadoop jar myjar.jar ..."

Posted 1 year ago #
deltafoxtrot
Member

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?

Posted 1 year ago #
tjake
Moderator

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...

Posted 1 year ago #
deltafoxtrot
Member

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

Posted 1 year ago #
deltafoxtrot
Member

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;
}

}

Posted 1 year ago #
tjake
Moderator

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 :)

Posted 1 year ago #
deltafoxtrot
Member

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)

Posted 1 year ago #

RSS feed for this topic

Reply

You must log in to post.