AvroRowDeserializationSchema extends AbstractDeserializationSchema<Row> { @Override public Row deserialize(byte[] message) throws IOException { try { inputStream.setBuffer(message); this.record = datumReader.read(record, decoder); } catch (IOException e) { throw new RuntimeException("Failed to deserialize Row.", e); } final Object row = convertToRow(schema, record); return (Row) row; } AvroRowDeserializationSchema(Class<? extends SpecificRecord> recordClazz); @Override Row deserialize(byte[] message);  }
@Test public void testSerializeDeserializeSimpleRow() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); final byte[] bytes = serializationSchema.serialize(testData.f2); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testSerializeSimpleRowSeveralTimes() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); serializationSchema.serialize(testData.f2); serializationSchema.serialize(testData.f2); final byte[] bytes = serializationSchema.serialize(testData.f2); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testDeserializeRowSeveralTimes() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); final byte[] bytes = serializationSchema.serialize(testData.f2); deserializationSchema.deserialize(bytes); deserializationSchema.deserialize(bytes); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testSerializeDeserializeComplexRow() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); final byte[] bytes = serializationSchema.serialize(testData.f2); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testSerializeComplexRowSeveralTimes() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); serializationSchema.serialize(testData.f2); serializationSchema.serialize(testData.f2); final byte[] bytes = serializationSchema.serialize(testData.f2); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testDeserializeComplexRowSeveralTimes() throws IOException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData(); final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0); final byte[] bytes = serializationSchema.serialize(testData.f2); deserializationSchema.deserialize(bytes); deserializationSchema.deserialize(bytes); final Row actual = deserializationSchema.deserialize(bytes); assertEquals(testData.f2, actual); }
@Test public void testSerializability() throws IOException, ClassNotFoundException { final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData(); final AvroRowSerializationSchema serOrig = new AvroRowSerializationSchema(testData.f0); final AvroRowDeserializationSchema deserOrig = new AvroRowDeserializationSchema(testData.f0); byte[] serBytes = InstantiationUtil.serializeObject(serOrig); byte[] deserBytes = InstantiationUtil.serializeObject(deserOrig); AvroRowSerializationSchema serCopy = InstantiationUtil.deserializeObject(serBytes, Thread.currentThread().getContextClassLoader()); AvroRowDeserializationSchema deserCopy = InstantiationUtil.deserializeObject(deserBytes, Thread.currentThread().getContextClassLoader()); final byte[] bytes = serCopy.serialize(testData.f2); deserCopy.deserialize(bytes); deserCopy.deserialize(bytes); final Row actual = deserCopy.deserialize(bytes); assertEquals(testData.f2, actual); }