Electric Boogaloo

This seems like the sort of thing somebody would want to do. I mean, I wanted to do it. I had some JSON messages and wanted to write them to HDFS as Avro to query them in Hive. The thing is it’s not exactly obvious how one might do such a thing in nifi. Or at least it wasn’t obvious to me, so this guide will attempt to demystify all that. It won’t demystify Avro itself, that is left as an exercise to the reader. FULL DISCLOSURE: This is my first blog post, ever.

We’ll start by spinning up a Docker container for nifi:

docker run -it --name nifi -p 8080:8080 apache/nifi

Then in your browser, go to localhost:8080 and the nifi interface should come up. It may take a minute. Your patience will be rewarded.

The first thing we’ll need is some JSON data to turn into Avro. The canonical example of this is to consume the Twitter sample endpoint, but (a) requires getting a developer account and (b) that payload is way bigger than I want to hassle with. So instead let’s drop a Generate Flowfile processor on the canvas and use it to generate some generic data. Open up the Configuration and set the custom text to some sort of valid JSON:

{"foo": "sample data", "bar": "more sample data"}

UpdateAttribute

Before we can convert the JSON to Avro we need to inform nifi of the Avro schema, which we will do with an UpdateAttribute step. The Avro schema will be carried along with the FlowFiles as an attribute for use downstream. In order to add the schema as an attribute, we will add a custom property through the UpdateAttribute processor configuration. Click the [+] to bring up the Add Property modal.

The property name we’re creating will be called avro.schema and the value will be the Avro schema for your JSON. Depending on the complexity of your data, you may be able to use an online JSON-Avro Schema Generator. It is possible for nifi to infer the Avro schema, but it will be blissfully unaware of things like required fields or any other complexities in your schema.

{
  "name": "MyClass",
  "type": "record",
  "namespace": "com.acme.avro",
  "fields": [
    {
      "name": "foo",
      "type": "string"
    },
    {
      "name": "bar",
      "type": "string"
    }
  ]
}

Once you have the Avro schema in hand, add the property, and paste your schema into the value field for that new property.

ConvertRecord

Now we can actually perform the conversion to Avro, using a ConvertRecord step. Nifi handles conversions like this with a Record Reader & Record Writer, each of which are controlled by controller services. For our purposes, we’ll need to create a new reader service using JsonTreeReader 1.10.0. When prompted, give it a descriptive name because nifi doesn’t do a particularly good job of allowing these services to be organized once you get a lot of them.

Once you create the controller service, you’ll need to configure it by click on that arrow next to the Record Reader

This will take you to the Controller Services page, where you can configure the service you just created by clicking on the gear icon.

Set the Schema Access Strategy to Use ‘Schema Text’ Property. Note that the Schema Text property has a value of ${avro.schema}. This means that the reader service will get the schema text from a FlowFile attribute named avro.schema. Hey, that’s what we named the property in our UpdateAttribute step! What an amazing coincidence. Click apply, and then click the lightning bolt to enable the service.

Next we’ll have to create and configure a Record Writer controller service. The steps are the same as the reader service, except that we want AvroRecordSetWriter 1.10.0, configured as such:

At this point if you run your flow, the output from the ConvertRecord step should look something like this, which is the Avro transformed message. You can see that the Avro schema is embedded in the message, which is because that’s what we told nifi to do in the record writer service.

Objavro.schemaŽ{"type":"record","name":"MyClass","namespace":"com.acme.avro","fields":[{"name":"foo","type":"string"},{"name":"bar","type":"string"}]}avro.codecnull�ÉŠKRïc
/B§k‹(Æ“:sample data more sample dataÉŠKRïc
/B§k‹(Æ“

PutHDFS

Now that we have our message in Avro format, let’s write it to HDFS with a PutHDFS step. You’ll need the hdfs-site.xml and core-site.xml files off your Hadoop cluster copied to your running Docker container, and you’ll need a folder created in HDFS to write the files to. The PutHDFS config should look something like this:

Query Avro data in Hive

By starting all our processes we will soon have data written to HDFS in Avro format, which you can see in the Hue file browser or with an hdfs dfs -ls from the command line. So what? You probably want to query that data, which is most easily done by creating a Hive external table. The Hive external table needs 3 things in order to work:

1. Serializer/de-serializer (aka SerDe)
2. Location of the avro data
3. Location of the avro schema

The AvroSerDe comes installed with Cloudera CDH/CDP. The Avro data is in the location specified in your PutHDFS step in nifi. The location of the Avro schema is the one thing we’re missing. Perhaps the easiest way to solve this is to create a new file via Hue file browser one directory level above your avro data. The location doesn’t really matter, but you don’t want it to be co-located with your avro data files. Create the file, call it example_avro.avsc, and then paste the Avro schema text into the file.

CREATE EXTERNAL TABLE example_avro_hive_ext
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED as AVRO
LOCATION '/user/nifi/example_avro_data/' 
TBLPROPERTIES ('avro.schema.url'='hdfs:///user/nifi/example_avro.avsc');

And just like that you can query your Avro data through Hive.

But what about the small file problem?

The astute observer will no doubt notice that as built, our pipeline will write one file in HDFS for each FlowFile, which means one per generated message. That’s fine for our example, but HDFS doesn’t like tons of tiny files which is exactly what we would have if we were consuming live sensor data or the Twitter firehose or whatever. The solution to this is to introduce a MergeContent step before our ConvertRecord step. MergeContent simply takes multiple FlowFiles and merges them into a single FlowFile. Doing it in this sequence puts one instance of the avro schema embedded in each file whereas if we merged content after the ConvertRecord, each record in the FlowFile would have the avro schema embedded. It turns out Hive doesn’t like that so much.

This whole task is rather straightforward in hindsight, and now that I understand it, but it was tough sledding early on. A worthwhile exercise is to take the converted Avro record in nifi and Convert it back into JSON with another set of record reader/writer controller services. Let me know if you found this helpful, I know I wish something like this existed when I was getting started with nifi.

In a future post we will examine the Schema Registry capabilities within CDP, and how it can simplify much of what we built out above.

Useful links:

Generate Avro schema from JSON data

Validate Avro schema

Avro tools

https://repo1.maven.org/maven2/org/apache/avro/avro-tools/1.9.1/

These are good for testing avro from the command line.

java -jar avro-tools-1.9.1.jar