Sorry I found that my previous message in the archive become all in black. Let
me re-explain the problem. The following piece of code for AvroReducer causes
problem:
public void reduce(Utf8 key, Iterable<GenericRecord> values,
AvroCollector<GenericRecord> collector, Reporter reporter) throws IOException {
GenericRecord record = null;
for (GenericRecord value : values) { -- code
omitted here -- record = value;
record.put("rowkey", key); <=== this statement causes problem
collector.collect(record);
} }
As explained in my previous message, if I remove the statement
record.put("rowkey", key), the code works fine, in that the key values pairs
passed to the routine reduce() are correct. But if you add this statement, the
key values pairs passed to the routine reduce() are out of order, something
like (key1, values1), (key2, values3) rather than (key2, values2). Some
details are explained in my previous message. Is this problem relating to
Hadoop binary iterators or Avro deserialization code? Thanks.
Ey-Chih Chow
From: [email protected]
To: [email protected]
Subject: is this a bug?
Date: Wed, 2 Mar 2011 13:05:55 -0800
Hi,
I am working on an Avro MR job and encountering an issue with AvroReducer<Utf8,
GenericRecord, GenericRecord>. The corresponding reduce() routine is
implemented in the following way:
public void reduce(Utf8 key, Iterable<GenericRecord> values,
AvroCollector<GenericRecord> collector, Reporter reporter) throws IOException {
. .
.
GenericRecord record = null;
for (GenericRecord value : values) { .
. .
record = value; record.put("rowkey", key);
. .
. collector.collect(record); }}
If I comment out the statement in red in the above code, the reduce function
gets called properly with CORRECT key values pairs passed to reduce().
However, if I add the statement in red to the routine, the reduce function is
called with WRONG key values pairs, in the sense that key2 paired with values3,
instead of values2, when passed to the reduce() routine. I traced this problem
by including Hadoop source code, such as ReduceTask.java, Task.java, and Avro
source code, such as HadoopReducer.java, HadoopReducerBase.java, and all the
serialization code. The problem showed up on the second call of the reduce(),
but I can not locate the exact place that cause the problem. My intuition is
that this is incurred in either the hadoop iterators after merge sort or Avro
deserialization. Is there anybody can help me on this? Thanks.
Ey-Chih Chow
