Had to do a lot of bisecting to finally get to where the issue seems to be,
going to take some time to get a separate junk of code that I can share out but
let me try to talk through the problem, maybe it's obvious.
I have raw text data that gets parsed into Log avro classes, which then go
through another DoFn to get converted into Visit avros classes and I want to
write out both collections of Log and Visit records. Basically the issue seems
to happen in the DoFn that takes in Log and creates Visit. If I try to set any
of the fields in Visit using fields from Log then I end up writing out no
files, if I set the fields in Visit with literals then I have output.
Here's the code, I thought this might be a detached values issue and tried that
as well but it didn't seem to help. Originally the code just used "l" directly
in process() and I didn't bother with "l2".
public static PTable<Visit, Pair<Long, Long>> parseVisit(PCollection<Log> logs)
{
return logs.parallelDo("second-parse", new DoFn<Log,
Pair<Visit,Pair<Long, Long>>>() {
private PType<Log> ptype;
@Override
public void initialize() {
this.ptype = Avros.specifics(Log.class);
this.ptype.initialize(getConfiguration());
}
@Override
public void process(Log l, Emitter<Pair<Visit, Pair<Long,Long>>>
emitter) {
Log l2 = ptype.getDetachedValue(l);
increment("test","visitcount");
// This works
Visit v = new Visit();
v.setDate(1234);
v.setDatehour("123");
v.setUser("123");
emitter.emit(Pair.of(v, Pair.of(1L, 1L)));
// This doesn't work
//Visit v = new Visit();
//v.setDate(l2.getDate());
//v.setUser(l2.getUser());
//emitter.emit(Pair.of(v, Pair.of(1L, 1L)));
}
}, Avros.tableOf(Avros.specifics(Visit.class), Avros.pairs(Avros.longs(),
Avros.longs())));
}
Thanks!
From: [email protected]
Date: Fri, 22 Aug 2014 08:57:13 -0700
Subject: Re: Multiple Writes in a single pipeline.
To: [email protected]
Many thanks for taking the time, would like to get this resolved before the
next release.
J
On Fri, Aug 22, 2014 at 8:55 AM, Danny Morgan <[email protected]> wrote:
Hi Josh,
The Log and Visits classes are all in the same jar, the classloader fix is in
place but I still get the issue without setting the class loader.
I'll put together the smallest reproduction I can and send out the code.
Danny
From: [email protected]
Date: Fri, 22 Aug 2014 08:42:06 -0700
Subject: Re: Multiple Writes in a single pipeline.
To: [email protected]
Hey Danny,
I wrote the test that I inlined below and it ran successfully for me against
master and the 0.10 branch, so there must be something more subtle going on
here-- are the Log and Visit classes created in different jars? I'm assuming
the classloader fix is in play here and I'm wondering if there is something
weird there.
import org.apache.crunch.MapFn;import org.apache.crunch.PCollection;import
org.apache.crunch.Pipeline;import org.apache.crunch.impl.mr.MRPipeline;import
org.apache.crunch.io.From;
import org.apache.crunch.io.To;import org.apache.crunch.test.Employee;import
org.apache.crunch.test.Person;import org.apache.crunch.test.TemporaryPath;
import org.apache.crunch.test.TemporaryPaths;import
org.apache.crunch.types.avro.Avros;import
org.apache.hadoop.fs.FileStatus;import org.apache.hadoop.fs.FileSystem;import
org.apache.hadoop.fs.Path;
import org.junit.Rule;import org.junit.Test;
public class MultiAvroOutputIT {
@Rule public transient TemporaryPath tmpDir = TemporaryPaths.create();
@Test public void testMultiAvroWrite() throws Exception { Path person =
tmpDir.getPath("person"); Path employee = tmpDir.getPath("employee");
Pipeline p = new MRPipeline(MultiAvroOutputIT.class,
tmpDir.getDefaultConfiguration()); PCollection<String> shakes =
p.read(From.textFile(tmpDir.copyResourcePath("shakes.txt")));
shakes.parallelDo(new PersonFn(), Avros.specifics(Person.class))
.write(To.avroFile(person)); shakes.parallelDo(new EmployeeFn(),
Avros.specifics(Employee.class))
.write(To.avroFile(employee)); p.run();
FileSystem fs = FileSystem.get(tmpDir.getDefaultConfiguration());
System.out.println("Person");
for (FileStatus fstat : fs.listStatus(person)) {
System.out.println(fstat.getPath() + ": " + fstat.getLen()); }
System.out.println("Employee");
for (FileStatus fstat : fs.listStatus(employee)) {
System.out.println(fstat.getPath() + ": " + fstat.getLen()); }
p.done(); }
static class PersonFn extends MapFn<String, Person> { @Override public
Person map(String input) { return new Person(); } }
static class EmployeeFn extends MapFn<String, Employee> { @Override
public Employee map(String input) { return new Employee(); }
}
}
On Fri, Aug 22, 2014 at 8:12 AM, Josh Wills <[email protected]> wrote:
That is super-interesting; let me try to replicate it in a test.
J
On Fri, Aug 22, 2014 at 7:26 AM, Danny Morgan <[email protected]> wrote:
This issue looks similar to https://issues.apache.org/jira/browse/CRUNCH-67
It turns out even if I get rid of the reduce phase and do just this:
PTable<String, String> lines = this.read(mySource);
PCollection<Log> parsed = lines.parallelDo("initial-parsing", new myParser(),
Avros.specifics(Log.class));
PTable<Visit, Pair<Long, Long>> visits = parsed.parallelDo("visits-parsing",
new VisitsExtractor(),
Avros.tableOf(Avros.specifics(Visit.class),
Avros.pairs(Avros.longs(), Avros.longs())));
visits.write(To.avroFile(outputPath+"/visits"), WriteMode.OVERWRITE);
parsed.write(To.avroFile(outputPath+"/raw"), WriteMode.OVERWRITE);
this.done();
The plan shows I should be writing to two different targets in a single map
phase however only "/raw" as data written out to it and "/visits" just contains
a _SUCCESS file and no data.
Might this be an issue writing out to two different Avro types in the same
phase?
Thanks Again,
Danny
From: [email protected]
To: [email protected]
Subject: RE: Multiple Writes in a single pipeline.
Date: Fri, 22 Aug 2014 02:02:20 +0000
Hi Josh,
From: [email protected]
Date: Thu, 21 Aug 2014 17:40:25 -0700
Subject: Re: Multiple Writes in a single pipeline.
To: [email protected]
The two different executions you have are doing different things, however. In
the first one, Crunch is running a single MapReduce job where the /raw
directory is written as a mapper side-output, and the /visits directory is
being written out on the reduce side (or at least, should be-- is there any
evidence of a failure in the job in the logs? Are bytes being written out from
the reducer?)
No evidence of any failures in the logs, the single mapper and reducers both
succeed. The mapper definitely writes to HDFS the reducer does not, here are
the relevant counters from the reducer:
FILE: Number of bytes read
6
FILE: Number of bytes written
91811
FILE: Number of large read operations
0FILE: Number of read operations
0
FILE: Number of write operations
0
HDFS: Number of bytes read
6205
HDFS: Number of bytes written
0
HDFS: Number of large read operations
0
HDFS: Number of read operations
4HDFS: Number of write operations
2
I couldn't find anything related on the crunch jira.
For this problem, I think it would be more efficient to write the parsed ->
/raw output first, call run(), then do the agg -> /visits output followed by
done(), which would mean that you would only need to parse the raw input once,
instead of twice.
Would the first option be more efficient if it worked?
A helpful trick for seeing how the Crunch planner is mapping your logic into
MapReduce jobs is to look at the plan dot file via one of the following
mechanisms:
1) Instead of calling Pipeline.run(), call Pipeline.runAsync() and then call
the getPlanDotFile() method on the returned PipelineExecution object. You can
print the dot file to a file and use a dot file viewer to look at how the DoFns
are broken up into MR jobs and map/reduce phases.
2) Call MRPipeline.plan() directly, which returns a MRExecutor object that also
implements PipelineExecution. (The difference being that calling
MRPipeline.plan will not start the jobs running, whereas calling runAsync will.)
I ran the two different version through dot and you're right they are two
complete different executions, pretty cool!
Thanks!
--
Director of Data ScienceClouderaTwitter: @josh_wills
--
Director of Data ScienceClouderaTwitter: @josh_wills
