DEV Community: Márton Papp

Beginners Guide to Caching Inside an Apache Beam Dataflow Streaming Pipeline Using Python

Márton Papp — Wed, 09 Mar 2022 12:06:05 +0000

Some days you win. Other days you have to process a continuous stream with thousands of incoming events per second from a PubSub subscription as quickly as possible. Even if your transformations are simple it is very likely that your business logic depends on some external and dynamic data. Let me show you a way how to efficiently cache these values and save you from hours of debugging by uncovering a bug in the Python library of Apache Beam itself.

If your configuration comes from an environment variable that changes once a year you can get away with updating your Dataflow every time but keep in mind that is usually takes a lot of time to execute by draining your existing pipelines and starting new ones.

If it’s more dynamic and can change every minute or so - for example you want to filter messages based on a criteria that a user can configure through UI - instead of using an env variable you will probably fetch a value from an external resource (e.g. a file from a bucket, a database query or an API call, you name it...) that you want to cache for a certain amount of time to prevent hammering your attached resource.

The naive implementation

Pipelines in Apache Beam are self-contained and isolated from each other, so they can be safely executed concurrently. But it also means that they can’t rely on shared state.

Depending on the load the Dataflow runner may start 1 to 10 worker processes and those workers can spawn hundreds or thousands of threads independent from each other. With a naive implementation each thread will use their own cache pool so even with a TTL high as 60 sec you can end up with ~10.000 API calls per minute. While in certain cases it’s better than making a call for every PubSub message it’s still far from ideal.

Shared handle in Python

Shared is a serializable object that can be shared by all threads of each worker process, and an instance of it returns Shared handle that can be used in your transform functions. This handle encapsulates a weak reference to a singleton object which will be our shared cache in form of a dict and can be accessed through the handle’s acquire function.

Please note, that most built-in types in Python cannot be weak referenced, but a dict can be easily made to support it through subclassing:

class _SideInputContainer(dict):
    pass

Watch out for a bug in Apache Beam versions <= 2.35.0

According to the documentation the acquire function accepts a constructor_fn and a tag parameter. The first one defines how you want to load your data on a cache miss and the tag defines an identifier you want use in subsequent calls to access the cached object.

Whenever the tag changes the cached data will be reloaded, which can ultimately serve as our TTL by writing a function that returns a different identifier every 60 seconds:

CONFIG_TTL_IN_SECONDS = 60
# Dummy (but usable!) implementation :)
def get_cache_key():
    return str(int(time.time() / CONFIG_TTL_IN_SECONDS))

You might think that writing

return shared_handle.acquire(load_sideinput, get_cache_key())

will do the job, but due to a bug in versions prior to this commit the tag parameter will be ignored. The cached object is going to be reloaded even if you provide the same identifier, rendering the whole mechanism useless and our transformation will hit our attached resources every time.

The workaround

The forward and backward compatible solution is to store this identifier on the cached dict itself, manually check if a reload is needed and call acquire again with a new tag.

class _SideInputContainer(dict):
    pass

def get_sideinput(shared_handle):
    def get_cache_key():
        return str(int(time.time() / 60))

    def load_sideinput():
        sideinput_container = _SideInputContainer()
        rawJson = # API call that returns a JSON that you want to cache
        sideinput_container["cached_response"] = json.loads(rawJson)
        sideinput_container["cache_key"] = get_cache_key()
        return sideinput_container

    sideinput_data = shared_handle.acquire(load_sideinput)
    current_cache_key = get_cache_key()
    if sideinput_data["cache_key"] != current_cache_key:
        sideinput_data = shared_handle.acquire(load_sideinput, current_cache_key)
    return sideinput_data

As always your likes and feedbacks are very much appreciated!

Push the button - Getting Started With IoT & MQTT

Márton Papp — Mon, 06 Apr 2020 10:04:48 +0000

Yes, I know what you think: there are already a ton of MQTT related posts and videos circulating around the net. But here's a thing: most of them either failed to demonstrate how to get started with an MQTT broker, or did not show any code about how to get things working, and did not provide a handy initial concept.

My goal with this series is to teach you how to publish events from a simple IoT button using MQTT, and then receive and react to button presses on two separate devices at the same time - a computer with NodeJS and an another connected Arduino device. All this as simple as possible, even without signing up to a hosted message broker like CloudMQTT.

Sounds good? Now let's get into it!

Setup your own MQTT broker

Assuming you have docker installed, setting up a local MQTT broker is much simpler than you might expect. Just execute the following command:

docker run -it -p 1883:1883 -p 9001:9001 eclipse-mosquitto

This tells docker to download the latest image of the open-source MQTT broker Eclipse Mosquitto, and start a container based on that image.

BOOM, now you have fully-functioning MQTT server listening on port 1883. Later you can run the container in the background with the --detach or -d option, but this way you can see the logs more easily. You can stop the container any time using CTRL+C.

Take a look behind the curtain

Unlike RabbitMQ, Mosquitto doesn't have a built-in management plugin with a UI, but without one it's hard to see what's happening under the hood. Fortunately there are multiple brilliant solutions out there, the best ones are MQTT.fx and MQTT Explorer, and the latter will be used throughout this tutorial.

It's a good idea to figure out your local IP address to use instead of localhost, because later the connected Arduino devices will need to know the broker IP address and thats obviously not localhost from their point of view.

One note here: this is for learning purposes only, later with port forwarding you can expose your MQTT broker to the internet if you want to (maybe from an ever-connected Raspberry PI), but when you done experimenting I suggest you to use a Cloud MQTT provider.

After figuring out your IP address open up MQTT Explorer and try to connect to the broker. My IP is 192.168.0.195, so for me it looked something like this:

After clicking connect you should see the empty broker. Going back to the terminal where the container runs will be logs from the incoming connection from the client.

What's next?

In the next chapter i'll show you how to publish messages to the broker using C/C++ code with M5Stick-C, an inexpensive ESP32 powered mini IoT device and consuming those messages in NodeJS.

Knex ❤️ PSQL: updating timestamps like a pro

Márton Papp — Tue, 25 Feb 2020 14:22:51 +0000

Knex.js is the most popular SQL query builder around and the go-to solution for most of us working with PostgreSQL. You can find dozens of articles on dev.to about how to get started, so I decided to focus on a more advanced and often overlooked topic on how to keep the updated_at fields really updated - automatically.

What does table.timestamps() do?

If you read along the documentation, upon creating a new table you will probably write a migration like this:

exports.up = function(knex) {
  return knex.schema.createTable('products', function(table) {
    table.increments('id').primary();
    table.string('name');
    table.timestamps(false, true);
  });
};

The table.timestamps(false, true) line adds created_at and updated_at columns on the table. Both columns default to being not null and using the current timestamp when true is passed as the second argument.
While it's sufficent for the created_at column, the updated_at will remain unchanged even after an update query executed and it's your responsibility to keep it in sync.

There is a good reason behind this behaviour: different SQL dialects - like MySQL - handle automatic updating pretty well, but others, like PostgreSQL don't support it.

What is a Trigger Procedure in PSQL?

Think of a trigger procedures like middlewares in expressjs. You have the opportunity to execute functions that modify the inserted values before actually committing the update. The NEW value holds the new database row for INSERT/UPDATE operations, so setting the updated_at field is really easy:

BEGIN
    NEW.updated_at = CURRENT_TIMESTAMP;
    RETURN NEW;
END;

Okay, okay, just give me the code already

First you need to create this trigger function in a migration using knex.raw:

exports.up = function(knex) {
  return knex.raw(`
    CREATE OR REPLACE FUNCTION update_timestamp() RETURNS TRIGGER
    LANGUAGE plpgsql
    AS
    $$
    BEGIN
        NEW.updated_at = CURRENT_TIMESTAMP;
        RETURN NEW;
    END;
    $$;
  `);
};

exports.down = function(knex) {
  return knex.raw(`
    DROP FUNCTION IF EXISTS update_timestamp() CASCADE;
  `);
};

To make sure everything went fine execute the following query:

SELECT routine_name, routine_definition
FROM information_schema.routines
WHERE routine_type='FUNCTION' AND specific_schema='public';

+------------------+---------------------------------------------+
| routine_name     | routine_definition                          |
|------------------+---------------------------------------------|
| update_timestamp |                                             |
|                  |     BEGIN                                   |
|                  |         NEW.updated_at = CURRENT_TIMESTAMP; |
|                  |         RETURN NEW;                         |
|                  |     END;                                    |
|                  |                                             |
+------------------+---------------------------------------------+

But how to use the function?

This function alone does nothing, we also need to tell the database engine where and when to use it. The best place for this the upcoming migrations where you create a new table - going back to my first example the code will be this:

const tableName = 'products';

exports.up = async function(knex) {
  await knex.schema.createTable(tableName, function(table) {
    table.increments('id').primary();
    table.string('name');
    table.timestamps(false, true);
  });

  await knex.raw(`
    CREATE TRIGGER update_timestamp
    BEFORE UPDATE
    ON ${tableName}
    FOR EACH ROW
    EXECUTE PROCEDURE update_timestamp();
  `);
};

exports.down = function(knex) {
  return knex.schema.dropTable(tableName);
};

If you run the \d products command, at the bottom of the table you will see that the trigger function will be executed on each row update on this table.

> \d products
+------------+--------------------------+--------------------------------------------------------+
| Column     | Type                     | Modifiers                                              |
|------------+--------------------------+--------------------------------------------------------|
| id         | integer                  |  not null default nextval('products_id_seq'::regclass) |
| name       | character varying(255)   |                                                        |
| created_at | timestamp with time zone |  not null default now()                                |
| updated_at | timestamp with time zone |  not null default now()                                |
+------------+--------------------------+--------------------------------------------------------+
Indexes:
    "products_pkey" PRIMARY KEY, btree (id)
Triggers:
    update_timestamp_on_products BEFORE UPDATE ON products FOR EACH ROW EXECUTE PROCEDURE update_timestamp()

As always, your likes and feedbacks are much appreciated!

Pipeline API 🔥 - the best way to handle stream errors that nobody tells you about

Márton Papp — Tue, 18 Feb 2020 14:54:57 +0000

... except the documentation on the millionth page, without any context, buried deeper than a superfluous dependency in your node_modules directory.

A little background

Streams are cruel and unpredictable, but usually you can copy-paste top rated answers from Stackoverflow for years without having full comprehension over them - a very important skill that most of us mastered during our career.

But one day you will be asked to transform and upload huge amounts of data from a database table to Google Storage and you will probably write something like this:

/// this is bad, please do not do this!
async streamFromDbToGcloudBucket(incomingStream) {
  const file = ...

  return new Promise((resolve, reject) => {
    incomingStream
      .pipe(file.createWriteStream())
      .on('error', function(err) {
        reject(err);
      })
      .on('finish', function() {
        resolve();
      });
  });
}

Wrapped in a promise, piping incoming stream to a gCloud file pretty neat, huh? After months in production things started to go south as we got inactivity alerts that sometimes the files are not uploaded hourly as expected.

The ugly

During debugging I stumbled upon the following lines in the storage lib from Google:

fs.createReadStream(pathString)
  .on('error', callback!)
  .pipe(newFile.createWriteStream(options))
  .on('error', callback!)

What? You need multiple .on('error', callback)'s in the same chain? Am I stupid for not knowing this? As it turns out you need to subscribe to error handlers on every stream, because pipe does not propagate errors like you would expect. It also means that you need to repeat this for every pipe you use.

Pipeline to the rescue

Fortunately Node 10 introduced the Pipeline API to alleviate such problems. Instead of using pipe, you can use pipeline(...streams, callback). It does pretty much the same, except that callback will be called when the pipeline is fully done, or an error occurred at some point. Let's see how it works:

const { pipeline } = require('stream');

pipeline(
  readableStream,
  writableStream,
  (err) => {
    if (err) {
      console.error('Pipeline failed.', err);
    } else {
      console.log('Pipeline succeeded.');
    }
  }
);

One more thing

You might notice that it's not wrapped in a promise. The good news is that pipeline is promisify-able (is this even a word?) as well, so you can write this:

const pipeline = util.promisify(stream.pipeline);

await pipeline(
  readableStream,
  writableStream
);

... and wrap it in a try-catch block.

In any case, I hope you the above useful, and as my first article ever, your likes and feedbacks are much appreciated!