Avro & Java: Record Parsing

This tutorial will guide you through how to convert json to avro and then back to json. I suggest you first read through the documentation on Avro to familiarize yourself with it. This tutorial assumes you have a maven project already setup and a resources folder.

POM:

Add Avro Dependency

 

 

 

 

Add Jackson Dependency

Avro Schema File:

Next you need to create the avro schema file in your resources folder. Name the file “schema.avsc”. The extension avsc is the Avro schema extension.

{
    "namespace": "test.avro",
    "type": "record",
    "name": "MY_NAME",
    "fields": [
        {"name": "name_1", "type": "int"},
        {"name": "name_2", "type": {"type": "array", "items": "float"}},
        {"name": "name_3", "type": "float"}
    ]
}

Json Record to Validate:

Next you need to create a json file that conforms to your schema you just made. Name the file “record.json” and put it in your resources folder. The contents can be whatever you want as long as it conforms to your schema above.

{ "name_1": 234, "name_2": [23.34,654.98], "name_3": 234.7}

It’s Avro Time:

Imports:

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

Conversion to Avro and Back:

private void run() throws IOException {
	//Get the schema and json record from resources
	final ClassLoader loader = getClass().getClassLoader();
	final File schemaFile = new File(loader.getResource("schema.avsc").getFile());
	final InputStream record = loader.getResourceAsStream("record.json");
	
	//Create avro schema
	final Schema schema = new Schema.Parser().parse(schemaFile);

	//Encode to avro
	final byte[] avro = encodeToAvro(schema, record);

	//Decode back to json
	final JsonNode node = decodeToJson(schema, avro);

	System.out.println(node);
	System.out.println("done");
}

/**
 * Encode json to avro
 * 
 * @param schema the schema the avro pertains to
 * @param record the data to convert to avro
 * @return the avro bytes
 * @throws IOException if decoding fails
 */
private byte[] encodeToAvro(Schema schema, InputStream record) throws IOException {
	final DatumReader<GenericData.Record> reader = new GenericDatumReader<>(schema);
	final DataInputStream din = new DataInputStream(record);
	final Decoder decoder = new DecoderFactory().jsonDecoder(schema, din);
	final Object datum = reader.read(null, decoder);
	final GenericDatumWriter<Object> writer = new GenericDatumWriter<>(schema);
	final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
	final Encoder encoder = new EncoderFactory().binaryEncoder(outputStream, null);
	writer.write(datum, encoder);
	encoder.flush();

	return outputStream.toByteArray();
}

/**
 * Decode avro back to json.
 * 
 * @param schema the schema the avro pertains to
 * @param avro the avro bytes
 * @return the json
 * @throws IOException if jackson fails
 */
private JsonNode decodeToJson(Schema schema, byte[] avro) throws IOException {
	final ObjectMapper mapper = new ObjectMapper();
	final DatumReader<GenericData.Record> reader = new GenericDatumReader<>(schema);
	final Decoder decoder = new DecoderFactory().binaryDecoder(avro, null);
	final JsonNode node = mapper.readTree(reader.read(null, decoder).toString());

	return node;
}

AWS: Java Kinesis Lambda Handler

This entry is part 2 of 5 in the series AWS & Java

If you want to write a Lambda for AWS in Java that connects to a Kinesis Stream. You need to have the handler.

Maven:

<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk</artifactId>
    <version>1.11.109</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.7.1</version>
</dependency>

This is the method that AWS Lambda will call. It will look similar to the one below.

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent.KinesisEventRecord;
import com.amazonaws.services.kinesis.model.Record;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.JsonNode;

public void kinesisRecordHandler(KinesisEvent kinesisEvent, Context context) {
	final String awsRequestId = context.getAwsRequestId();
	final int memoryLimitMb = context.getMemoryLimitInMB();
	final int remainingTimeInMillis = context.getRemainingTimeInMillis();

	for (final KinesisEventRecord kinesisRec : kinesisEvent.getRecords()) {
		final Record record = kinesisRec.getKinesis();

		//We get the kinesis data information
		final JsonNode recData = new ObjectMapper().readValue(record.getData().array(), JsonNode.class);

		final String bucketName = recData.get("bucket").asText();
		final String key = recData.get("key").asText();
	}
}

The thing to note when you setup you Lambda is how to setup the “Handler” field in the “Configuration” section on AWS. It is in the format “##PACKAGE##.##CLASS##::##METHOD##”.

Java: Jackson Json

When you want to work with JSON the package of choice is Jackson. I find it so useful and easy to use.

Maven:

<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-core</artifactId>
	<version>2.7.1</version>
</dependency>
<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-annotations</artifactId>
	<version>2.7.1</version>
</dependency>
<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-databind</artifactId>
	<version>2.7.1</version>
</dependency>

There are so many things that we can do with Jackson. It’s actually very exciting.

Let’s start with converting a json string to ObjectNode.

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

//We need an object mapper to help with the conversion
final ObjectMapper mapper = new ObjectMapper();
//Next we read the json string into ObjectNode
final ObjectNode node = (ObjectNode)mapper.readTree(jsonString);

//You can check if the objectnode has a key by
node.has("my_key")

//You can get the value by
node.get("my_key")

//To get the value as a string using 
.asText() at the end of the get

//To get as int
.asInt()

//To get as boolean
.asBoolean()

//To get as double
.asDouble()

//To get as Long
.asLong()

Next let’s convert a json string to JsonNode

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

//You need an object mapper to help with the conversion
final ObjectMapper mapper = new ObjectMapper();
//Next we read the json string into a jsonnode
final JsonNode node = mapper.readTree(jsonString);
//We can if you want to switch it to an object node fairly easily.
final ObjectNode objNode = (ObjectNode)node;

//Put value in the object node
objNode.put("my_key", 1);

//You can also set json node
objNode.set("my_key", myJsonNode);

You can also create a new object node by using JsonNodeFactory. See below.

import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;

//If you want to create an instance of ObjectNode
final ObjectNode objNode = JsonNodeFactory.instance.objectNode();

If you want to work with json array’s it’s also pretty straight forward. Again we work with JsonNodeFactory to declare it out.

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;

private ArrayNode arrNode = JsonNodeFactory.instance.arrayNode();

//We can then add to it
arrNode.add();

Jackson also has generators which allow us to create json on the fly.

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonProcessingException;

//It throws a "JsonProcessingException'

final JsonFactory jfactory = new JsonFactory();
ByteArrayOutputStream out=new ByteArrayOutputStream();

try (final JsonGenerator jg = jfactory.createGenerator(out, JsonEncoding.UTF8)) {
	jg.setCodec(objectMapper);

	jg.writeStartObject();

	jg.writeFieldName("my_key");
	jg.writeString("hi");

	jg.writeFieldName("next_key");
	jg.writeObject(my_obj);

	jg.writeFieldName("third_key");
	jg.writeTree(my_obj_node);

	jg.writeEndObject();
}

Jackson JsonViews are really cool in my opinion. We can have an object that has multiple properties and we can set a view to each one of these properties. Which means when we serialize the object we will only get those fields that are associated to that view and vise versa on the deserialize.

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonView;

//You can however many views we want to work with in our objects. They must all be classes. 
//If you extend a class then it will contain all properties of that view and the view it extends.
public class MyClassViews {
	public static class ViewA {}
	public static class ViewB extends ViewA {}
}

//Let's create an object for our serialization and deserialization
public class MyClass {
	private final int id;
	private final String value;

	//This constructor is what the deserializer would use.
	@JsonCreator
	public MyClass(@JsonProperty("id") int id, @JsonProperty("value") String value) {
	}

	@JsonView(MyClassViews.ViewA.class)
	@JsonProperty("id") //You don't have to put the name you could leave it as @JsonProperty
	public int getId() {
		return id;
	}

	@JsonView(MyClassViews.ViewB.class)
	@JsonProperty("value")
	public String getValue() {
		return value;
	}
}

I didn’t come up with this next part but it’s so awesome that I wanted to include it. Let’s say you had an object that was populated not using any views but the object does contain views that will eventually pass up to a UI but you only want to pass using a specific view. Then you can do the following.

import javax.ws.rs.core.StreamingOutput;
import com.fasterxml.jackson.databind.ObjectMapper;

final StreamingOutput jsonStream = new StreamingOutput() {
	@Override
	public void write(OutputStream out) throws IOException {
		final ObjectMapper mapper = new ObjectMapper();
		mapper.writerWithView(MyClassViews.ViewB.class).writeValue(out, MyClass);
	}
};

If you want to write the object to string using a view. All you need to do is

final String jsonString = new ObjectMapper().writerWithView(ViewB).writeValueAsString(MyClass);

If you then want to convert our new jsonString to the object using the view do the following:

new ObjectMapper().readerWithView(ViewB.class).forType(MyClass.class).readValue(jsonString);

If you want to convert a json string to object.

new ObjectMapper().readValue(myJsonString, MyClass.class)