@Test public void testBasicFunctionality() throws Exception { Schema.Parser parser = new Schema.Parser(); Schema schema = parser.parse(SCHEMA); Assert.assertTrue(schema != null); List<Schema.Field> fields = schema.getFields(); Assert.assertEquals(3, fields.size()); GenericRecord user1 = new GenericData.Record(schema); user1.put("name", "Root"); user1.put("favorite_number", 8); GenericRecord user2 = new GenericData.Record(schema); user2.put("name", "Ben"); user2.put("favorite_number", 7); user2.put("favorite_color", "red"); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); ByteArrayOutputStream out = new ByteArrayOutputStream(); JsonEncoder encoder = EncoderFactory.get().jsonEncoder(schema, out); datumWriter.write(user1, encoder); datumWriter.write(user2, encoder); encoder.flush(); out.close(); byte[] bytes = out.toByteArray(); JsonAvroDecoder decoder = new JsonAvroDecoder(schema); List<Row> rows = decoder.decode(bytes); Assert.assertEquals(2, rows.size()); Assert.assertEquals("Root", rows.get(0).getValue("name")); Assert.assertEquals("Ben", rows.get(1).getValue("name")); }
@Override public List<Row> decode(byte[] bytes) throws DecoderException { List<Row> rows = new ArrayList<>(); JsonDecoder decoder = null; ByteArrayInputStream in = new ByteArrayInputStream(bytes); try { decoder = DecoderFactory.get().jsonDecoder(getSchema(), in); while (true) { try { GenericRecord gRecord = getReader().read(null, decoder); List<Schema.Field> fields = getSchema().getFields(); Row r = new Row(); for (Schema.Field field : fields) { Object object = gRecord.get(field.name()); if (object instanceof Utf8) { Utf8 o = (Utf8) object; object = o.toString(); } else if (object instanceof Map || object instanceof List) { object = gson.toJson(object); } r.add(field.name(), object); } rows.add(r); } catch (EOFException e) { break; } } } catch (AvroTypeException e) { throw new DecoderException(e.getMessage()); } catch (IOException e) { throw new DecoderException("Issue create json decoder, verify the schema"); } finally { try { in.close(); } catch (IOException e) { } } return rows; }
JsonAvroDecoder extends AbstractAvroDecoder { @Override public List<Row> decode(byte[] bytes) throws DecoderException { List<Row> rows = new ArrayList<>(); JsonDecoder decoder = null; ByteArrayInputStream in = new ByteArrayInputStream(bytes); try { decoder = DecoderFactory.get().jsonDecoder(getSchema(), in); while (true) { try { GenericRecord gRecord = getReader().read(null, decoder); List<Schema.Field> fields = getSchema().getFields(); Row r = new Row(); for (Schema.Field field : fields) { Object object = gRecord.get(field.name()); if (object instanceof Utf8) { Utf8 o = (Utf8) object; object = o.toString(); } else if (object instanceof Map || object instanceof List) { object = gson.toJson(object); } r.add(field.name(), object); } rows.add(r); } catch (EOFException e) { break; } } } catch (AvroTypeException e) { throw new DecoderException(e.getMessage()); } catch (IOException e) { throw new DecoderException("Issue create json decoder, verify the schema"); } finally { try { in.close(); } catch (IOException e) { } } return rows; } }
JsonAvroDecoder extends AbstractAvroDecoder { @Override public List<Row> decode(byte[] bytes) throws DecoderException { List<Row> rows = new ArrayList<>(); JsonDecoder decoder = null; ByteArrayInputStream in = new ByteArrayInputStream(bytes); try { decoder = DecoderFactory.get().jsonDecoder(getSchema(), in); while (true) { try { GenericRecord gRecord = getReader().read(null, decoder); List<Schema.Field> fields = getSchema().getFields(); Row r = new Row(); for (Schema.Field field : fields) { Object object = gRecord.get(field.name()); if (object instanceof Utf8) { Utf8 o = (Utf8) object; object = o.toString(); } else if (object instanceof Map || object instanceof List) { object = gson.toJson(object); } r.add(field.name(), object); } rows.add(r); } catch (EOFException e) { break; } } } catch (AvroTypeException e) { throw new DecoderException(e.getMessage()); } catch (IOException e) { throw new DecoderException("Issue create json decoder, verify the schema"); } finally { try { in.close(); } catch (IOException e) { } } return rows; } JsonAvroDecoder(Schema schema); }
JsonAvroDecoder extends AbstractAvroDecoder { @Override public List<Row> decode(byte[] bytes) throws DecoderException { List<Row> rows = new ArrayList<>(); JsonDecoder decoder = null; ByteArrayInputStream in = new ByteArrayInputStream(bytes); try { decoder = DecoderFactory.get().jsonDecoder(getSchema(), in); while (true) { try { GenericRecord gRecord = getReader().read(null, decoder); List<Schema.Field> fields = getSchema().getFields(); Row r = new Row(); for (Schema.Field field : fields) { Object object = gRecord.get(field.name()); if (object instanceof Utf8) { Utf8 o = (Utf8) object; object = o.toString(); } else if (object instanceof Map || object instanceof List) { object = gson.toJson(object); } r.add(field.name(), object); } rows.add(r); } catch (EOFException e) { break; } } } catch (AvroTypeException e) { throw new DecoderException(e.getMessage()); } catch (IOException e) { throw new DecoderException("Issue create json decoder, verify the schema"); } finally { try { in.close(); } catch (IOException e) { } } return rows; } JsonAvroDecoder(Schema schema); @Override List<Row> decode(byte[] bytes); }
JsonAvroDecoder extends AbstractAvroDecoder { @Override public List<Row> decode(byte[] bytes) throws DecoderException { List<Row> rows = new ArrayList<>(); JsonDecoder decoder = null; ByteArrayInputStream in = new ByteArrayInputStream(bytes); try { decoder = DecoderFactory.get().jsonDecoder(getSchema(), in); while (true) { try { GenericRecord gRecord = getReader().read(null, decoder); List<Schema.Field> fields = getSchema().getFields(); Row r = new Row(); for (Schema.Field field : fields) { Object object = gRecord.get(field.name()); if (object instanceof Utf8) { Utf8 o = (Utf8) object; object = o.toString(); } else if (object instanceof Map || object instanceof List) { object = gson.toJson(object); } r.add(field.name(), object); } rows.add(r); } catch (EOFException e) { break; } } } catch (AvroTypeException e) { throw new DecoderException(e.getMessage()); } catch (IOException e) { throw new DecoderException("Issue create json decoder, verify the schema"); } finally { try { in.close(); } catch (IOException e) { } } return rows; } JsonAvroDecoder(Schema schema); @Override List<Row> decode(byte[] bytes);  }