What is a Heap?

A heap is a specialized data structure that is designed to always yield the smallest element in O(lg(n)) time (assuming it is a min-heap; a max-heap would do the opposite.

It is logically represented as a balanced binary tree with all the elements shifted as left as possible. Practically it is represented as an array because it is actually very easy to map a left-shifted binary tree into an array. The left child of any node is index 2 * N + 1 in the array and the right child of any node is index 2 * N + 2 of the array.

The O(lg(n)) retrieval time of a min or max element makes heaps ideal to be used as a priority queue.

How Does it Work?

The key to it providing this performance is that all elements are stored in such way that every element is guaranteed to be less than either of its child elements (so, logically this means the root is the smallest element). If it was a max-heap the opposite would be true.

It accomplishes this in a pretty simple way. Let’s say we’re using a min-heap:

• The smallest element will be at the root of the tree.
• So, to get the next smallest element, we just:
  • Take the root of the tree.
  • Move the last element in the tree (the rightmost leaf node)
    up to where the root element was.
  • Repeatedly switch this node with any nodes below it that are
    less than it (this is called “percolating down”.
  • Taking the root takes O(1) time, and percolating down at most
    takes the height of the tree, which for a binary tree is O(lg(n)).
    So, this is how we get O(lg(n)) time for taking each element.
• If we want to insert new elements, we basically do the opposite:
  • Add the new element at the last position of the tree.
  • Repeatedly switch it upwards if it is less than its parent.
  • This is called “percolating up” and also is O(log(n)) time since
    it is also dependent on the height of the tree.

Creating a Heap

Surprisingly, a heap can actually be constructed in linear time O(n) from an existing array of objects.

This is a little time consuming to prove, so please refer to the method here: https://www.geeksforgeeks.org/time-complexity-of-building-a-heap/.

Implementation

Here is a full sample implementation of a heap (minus the construction of a heap from an existing array; in this case we just build it up using add and remove).

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.NoSuchElementException;

public class MinHeap<T extends Comparable> {

    private List heap;

    public MinHeap() {
        heap = new ArrayList();
    }

    //Add to end, percolate up to correct position.
    public void add(T value) {
        heap.add(value);

        for (int index = heap.size() - 1; index > 0; index = index / 2) {
            int parent = index / 2;
            if (heap.get(index).compareTo(heap.get(parent)) < 0) {
                T temp = heap.get(index);
                heap.set(index, heap.get(parent));
                heap.set(parent, temp);
            }
        }
    }

    private static int getLeft(int i) { return i * 2 + 1; }
    private static int getRight(int i) { return i * 2 + 2; }

    //Take from top, move last element to root, percolate down.
    public T remove() {
        if (heap.size() == 0) throw new NoSuchElementException();
        T result = heap.get(0);

        //Remove the last element.  If this makes the heap empty,
        //then just return the result.
        T last = heap.remove(heap.size() - 1);
        if (heap.size() == 0) return result;

        //If we got here, the heap still has elements, so add the last
        //element to the root (0) and percolate down.  Percolating down
        //means swapping the root with its children recursively until
        //it is smaller than both of its children.
        heap.set(0, last);
        int i = 0;
        while (i = getRight(i)) {
                T leftChild  = heap.get(getLeft(i));
                T rightChild = heap.get(getRight(i));
                if (leftChild.compareTo(rightChild) > 0) {
                    minChildIndex = getRight(i);
                }
            }
            if (heap.get(i).compareTo(heap.get(minChildIndex)) > 0) {
                T temp = heap.get(i);
                heap.set(i, heap.get(minChildIndex));
                heap.set(minChildIndex, temp);
                i = minChildIndex;
            }
            else break;
        }

        return result;
    }

    public boolean isEmpty() { return heap.isEmpty(); }

    public static void main(String[] args) {
        MinHeap h = new MinHeap();
        Arrays.asList(7, 6, 2, 9, 8, 17, 46, 12, 13, 1, 7)
            .forEach(h::add);
        while(!h.isEmpty()) { System.out.print(h.remove() + " "); }
        System.out.println();

        Arrays.asList(7, 6, 2, 9, 8, 17, 46, 12, 13, 1, 7, 51)
            .forEach(h::add);
        while(!h.isEmpty()) { System.out.print(h.remove() + " "); }

        System.out.println();
        h.add(1);
        while(!h.isEmpty()) { System.out.print(h.remove() + " "); }
    }
}

Continue reading →

I was writing another blog post focusing on a hanging error in MapR-Streams, and I ended up going into some good detail on the basic architecture of the product.  So, I though it was worth breaking out into its own blog.

What is MapR-Streams?

MapR-Streams is basically MapR’s version of Apache Kafka.  It is built into their cluster automatically though; it is actually a part of their core file system (along with MapR-DB).

Apache Kafka and MapR-Streams are basically a newer form of message queuing service that is particularly well-geared for big data and extremely high loads (read: millions of messages a per second in many common use cases).

As we dive into more detail, you can refer to this diagram to get a mental picture of how the product is used.  It was taken from a MapR presentation which can be found right here.

MapR-Streams Basics

• You create a “stream” in MapR.
• A stream is a container for topics.
• A topic is basically a message queue broken into partitions.
• Each partition is a unit of parallelism… realistically, each one is basically a queue within the topic.
• Each partition is replicated to multiple servers for high availability.
• Messages within each partition are guaranteed to be in the order in which they were sent.

Programming Languages

MapR-Streams can be used with a variety of languages including Java, Python, and C++.  It also has a RESTFUL interface which would make it available from pretty much anywhere.

Writing Messages

You use the Kafka APIs to send messages to MapR-Streams topics.  There is some one-off code to configure the producer; but after that, it’s pretty must just 1-2 lines of code to send a message from then on.

When sending the messages, you can either:

1. Target the topic with no partitions specified.  This will cause messages you send to be distributed across all partitions in the topic.
2. Target specific partitions in a topic to send to.  For example, I could make sure that all metrics messages for Server A’s CPU metric went to partition #3.  This way, they would be guaranteed to be in order when read by any consumers.

Creating a Kafka Producer:

public static void configureProducer(String[] args) {
    Properties props = new Properties();
    props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
    props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
    producer = new KafkaProducer(props);
}

Sending a Sample Message:

ProducerRecord rec = new ProducerRecord(
topic, messageText);
producer.send(rec);

You can find a full producer example from MapR here (I just referenced code from it): https://mapr.com/docs/52/MapR_Streams/code_for_the_sample_java_producer.html.

Consuming Messages

When you want to consume messages from a topic, usually the Kafka subscribe API is used (although there is an alternative/older assign API too).

• When you subscribe to a topic, you provide a consumer-group name; it’s just an arbitrary string; you could call it “woohoo!” if you felt like it.
• All consumers that subscribe to the same topic with the same consumer-group will “share” the partitions of that topic.  By share, I mean that consumers will all get a subset of the partitions.  A partition is only ever owned by one consumer a time.
• MapR-Streams stores a long integer offset for each partition for each consumer group.  So, if you had 2 consumer-groups reading a topic with 4 partitions, MapR-Streams would be tracking 8 offsets.

As practical example… If I have 2 Java processes, each with 2 threads having 1 consumer each, then I have 4 consumers running.  If the consumers subscribe to a topic with 16 partitions, it is likely that each of the 4 consumes will end up with 4 partitions.

Again, only one consumer in a consumer-group can own a partition of a topic at a time.  So… if the consumer dies or is shut down, it will release the partition, and another consumer will automatically pick it up (which is totally awesome).  This means that you can have 4 servers reading the same topic in parallel and you can be confident that if one server dies, the other 3 will pick up its share of the load transparently without you doing anything on the client side.

Creating a Kafka Consmer:

public static void configureConsumer(String[] args) {
    Properties props = new Properties();
    props.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");

    consumer = new KafkaConsumer(props);
}

Consuming Messages:

List topics = new ArrayList();
topics.add(topic);
consumer.subscribe(topics);

//You should wrap this line in a loop so it happens forever, or until you
//have met some terminal condition.
ConsumerRecords consumerRecords = consumer.poll(
pollTimeout);

Iterator iterator =
consumerRecords.iterator();

while (iterator.hasNext()) {
    ConsumerRecord record = iterator.next();
    System.out.println((" Consumed Record: " +
    record.toString()));
}

You can find a full consumer example from MapR here (I just referenced code from it): https://mapr.com/docs/52/MapR_Streams/code_for_the_sample_java_consumer.html.