[
https://issues.apache.org/jira/browse/MRUNIT-197?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthew Hayes updated MRUNIT-197:
---------------------------------
Attachment: MemberEventCountUnitTest.java
> Problems using Avro with MRUnit
> -------------------------------
>
> Key: MRUNIT-197
> URL: https://issues.apache.org/jira/browse/MRUNIT-197
> Project: MRUnit
> Issue Type: Bug
> Affects Versions: 1.0.0
> Reporter: Matthew Hayes
> Attachments: MemberEventCountUnitTest.java
>
>
> I'm not able to use MRUnit with Avro in a particular use case. The error I
> get is:
> java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record
> cannot be cast to java.lang.Number
> at
> org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:78)
> at
> org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:104)
> at
> org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:58)
> at
> org.apache.avro.hadoop.io.AvroSerializer.serialize(AvroSerializer.java:104)
> at
> org.apache.avro.hadoop.io.AvroSerializer.serialize(AvroSerializer.java:46)
> at
> org.apache.hadoop.mrunit.internal.io.Serialization.copy(Serialization.java:74)
> at
> org.apache.hadoop.mrunit.internal.io.Serialization.copy(Serialization.java:91)
> at
> org.apache.hadoop.mrunit.internal.io.Serialization.copyWithConf(Serialization.java:104)
> at org.apache.hadoop.mrunit.TestDriver.copy(TestDriver.java:608)
> at org.apache.hadoop.mrunit.TestDriver.copyPair(TestDriver.java:612)
> at
> org.apache.hadoop.mrunit.MapDriverBase.addInput(MapDriverBase.java:118)
> at
> org.apache.hadoop.mrunit.MapDriverBase.withInput(MapDriverBase.java:207)
> at
> com.linkedin.hogwarts.examples.tests.MemberEventCountUnitTest.mapTest(MemberEventCountUnitTest.java:59)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
> at java.lang.reflect.Method.invoke(Method.java:597)
> at
> org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:84)
> at org.testng.internal.Invoker.invokeMethod(Invoker.java:714)
> at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:901)
> at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1231)
> at
> org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:127)
> at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:111)
> at org.testng.TestRunner.privateRun(TestRunner.java:767)
> at org.testng.TestRunner.run(TestRunner.java:617)
> at org.testng.SuiteRunner.runTest(SuiteRunner.java:334)
> at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:329)
> at org.testng.SuiteRunner.privateRun(SuiteRunner.java:291)
> at org.testng.SuiteRunner.run(SuiteRunner.java:240)
> at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
> at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:86)
> at org.testng.TestNG.runSuitesSequentially(TestNG.java:1224)
> at org.testng.TestNG.runSuitesLocally(TestNG.java:1149)
> at org.testng.TestNG.run(TestNG.java:1057)
> at org.testng.remote.RemoteTestNG.run(RemoteTestNG.java:111)
> at org.testng.remote.RemoteTestNG.initAndRun(RemoteTestNG.java:204)
> at org.testng.remote.RemoteTestNG.main(RemoteTestNG.java:175)
> Here's a code sample that demonstrates the problem:
> import java.io.IOException;
> import java.util.Arrays;
> import org.apache.avro.Schema;
> import org.apache.avro.Schema.Type;
> import org.apache.avro.generic.GenericData;
> import org.apache.avro.generic.GenericRecord;
> import org.apache.avro.mapred.AvroKey;
> import org.apache.avro.mapred.AvroValue;
> import org.apache.avro.mapreduce.AvroJob;
> import org.apache.avro.mapreduce.AvroKeyInputFormat;
> import org.apache.avro.mapreduce.AvroKeyOutputFormat;
> import org.apache.hadoop.io.NullWritable;
> import org.apache.hadoop.mapreduce.Job;
> import org.apache.hadoop.mapreduce.Mapper;
> import org.apache.hadoop.mrunit.mapreduce.MapDriver;
> import org.testng.annotations.Test;
> public class MemberEventCountUnitTest
> {
> public static Schema INPUT_SCHEMA;
>
> static
> {
> INPUT_SCHEMA = Schema.createRecord("MemberId", null, "com.test", false);
> INPUT_SCHEMA.setFields(Arrays.asList(new
> Schema.Field("memberId",Schema.create(Type.INT),null,null)));
> }
>
> /**
> * Fails with: java.lang.ClassCastException:
> org.apache.avro.generic.GenericData$Record cannot be cast to java.lang.Number
> */
> @Test
> public void mapTest() throws IOException
> {
>
> MapDriver<AvroKey<GenericRecord>,NullWritable,AvroKey<Integer>,AvroValue<Integer>>
> mapDriver;
>
> Job job = new Job();
>
> job.setInputFormatClass(AvroKeyInputFormat.class);
> job.setOutputFormatClass(AvroKeyOutputFormat.class);
>
> AvroJob.setInputKeySchema(job, INPUT_SCHEMA);
> AvroJob.setMapOutputKeySchema(job, Schema.create(Type.INT));
> AvroJob.setMapOutputValueSchema(job, Schema.create(Type.INT));
>
> MyMapper mapper = new MyMapper();
> mapDriver =
> MapDriver.newMapDriver(mapper).withConfiguration(job.getConfiguration());
>
> mapDriver.withInput(new AvroKey<GenericRecord>(createRecord(123)), null)
> .withOutput(new AvroKey<Integer>(123),new AvroValue<Integer>(1));
> mapDriver.runTest();
> }
>
> /**
> * Passes
> */
> @Test
> public void mapTest2() throws IOException
> {
>
> MapDriver<AvroKey<Integer>,NullWritable,AvroKey<Integer>,AvroValue<Integer>>
> mapDriver2;
>
> Job job = new Job();
>
> job.setInputFormatClass(AvroKeyInputFormat.class);
> job.setOutputFormatClass(AvroKeyOutputFormat.class);
>
> AvroJob.setInputKeySchema(job, Schema.create(Type.INT));
> AvroJob.setMapOutputKeySchema(job, Schema.create(Type.INT));
> AvroJob.setMapOutputValueSchema(job, Schema.create(Type.INT));
>
> MyMapper2 mapper2 = new MyMapper2();
> mapDriver2 =
> MapDriver.newMapDriver(mapper2).withConfiguration(job.getConfiguration());
>
> mapDriver2.withInput(new AvroKey<Integer>(123), NullWritable.get())
> .withOutput(new AvroKey<Integer>(123),new AvroValue<Integer>(1));
> mapDriver2.runTest();
> }
>
> /**
> * Map an input record to input to (memberId, 1).
> */
> public static class MyMapper extends
> Mapper<AvroKey<GenericRecord>,NullWritable,AvroKey<Integer>,AvroValue<Integer>>
> {
> @Override
> protected void map(AvroKey<GenericRecord> key, NullWritable value,
> Context context) throws java.io.IOException, java.lang.InterruptedException
> {
> Integer memberId =
> (Integer)((GenericRecord)key.datum()).get("memberId");
> context.write(new AvroKey<Integer>(memberId), new
> AvroValue<Integer>(1));
> }
> }
>
> /**
> * Map an integer to input to (memberId, 1).
> */
> public static class MyMapper2 extends
> Mapper<AvroKey<Integer>,NullWritable,AvroKey<Integer>,AvroValue<Integer>>
> {
> @Override
> protected void map(AvroKey<Integer> key, NullWritable value, Context
> context) throws java.io.IOException, java.lang.InterruptedException
> {
> context.write(key, new AvroValue<Integer>(1));
> }
> }
>
> public static GenericRecord createRecord(int memberId)
> {
> GenericRecord record = new GenericData.Record(INPUT_SCHEMA);
> record.put("memberId", memberId);
> return record;
> }
> }
--
This message was sent by Atlassian JIRA
(v6.1.5#6160)
