4 ms·

Kafka has a different data model which works for some scenarios, but not others. We attempted to use Kafka as part of a job management system, where long-runni

by Jasper_ 11y ago

Kafka has a different data model which works for some scenarios, but not others. We attempted to use Kafka as part of a job management system, where long-running jobs were scheduled and workers consumed partitions, but what we found was that since consumers work on partitions, a long-running task could block an entire partition's worth of work, with no way to migrate it to another partition. Kafka works really well when the bottleneck is the broker to begin with -- if you have a lot of small, lightweight messages being passed to other systems, and the broker is having trouble keeping up. Analytics and logs are great examples where Kafka's data model works really well. But for a smaller number of larger messages that takes seconds to minutes to process (we were using it to schedule data downloads and video processing), workers could be sitting idle while there was still plenty more work to do. We moved to Amazon SQS for now, but I imagine we'll move to something like Disque in the future.

lobster_johnson 11y ago
I would argue that job management is unrelated to messaging, at least according my loose definition of a task that: * Has a well-defined lifetime — unstarted, running, paused, completed successfully, or failed; * Is executed from some kind of parameterized "job specification" that describes its inputs and desired behavior; * Has state data (e.g., completion % progress, log output, metrics, transactional continutation state for pausing/retrying); * Can be scheduled on multiple nodes; * Can be created and scheduled multiple times; * Can be scheduled by priority; * Can have its resources (RAM, CPU, I/O) constrained. The thing is, queues are terrible at this. People continue to abuse messaging systems as a scheduler, and it always sucks. For example, consider acking: For a long-running tasks that might take hours, do you really want to hold up the broker's internal lock for all that time? What if you restart the broker — the task is still running, now you decoupled its running state from the queue. Similarly, queues are terrible at finding out what's waiting, or what has happened. By using the queue as job state, you are coupling scheduling with execution, which is just wrong. A queue (one that supports priorities, mind you) is, however, excellent for management commands: "Start job X", "pause job X" and so on. But you will want to maintain execution state separately from the queue. A relational database is quite good at this, though one could use a NoSQL databases, too. First, create a row representing the job: {id="job1", state="new", type="import_stuff", params="..."}. Then push a queue message {commmand="start_job", job_id="job1"}. Run a queue consumer or ten on each node. Each consumer acks the message, finds the job in the table and starts it. One could run it as a child process, or in a Docker container. One could of course use an orchestration service (like Mesos/Marathon) to run it, and monitor the service for container events and update the job table appropriately. I would want to take it further and allow the job to update itself through API, so that it could publish metrics and log messages, as well as maintain whatever data is useful snapshotting. For example, a long-running task to import data into a database would benefit from maintaining a cursor so that you could pause/kill the job and later resume it at the point where it was.
- Jasper_ 11y ago
  Sure, that's actually very similar to the the system we built internally. But I don't see how Kafka as a messaging layer helps in your system, either. Presumably, the consumers won't advance their offset in their partition until the job is done, because they are not ready to start a new job. So you will still get the same blocking behavior. In addition, I don't see how you implement priority in a system like this, since Kafka partitions are append-only. One has to process the former start_job messages before you reach the higher priority items. Having a broker hand out items in priority to workers makes a lot more sense to me. The job system you described is definitely something missing from the OSS world, and we'd love to open up our system for this eventually. While Disque is not that piece, I think it makes more sense than Kafka.
  - lobster_johnson 11y ago
    I didn't mean to imply that Kafka was suitable for such a system, though that's mostly because it doesn't support priority queues. You could work around this by using a fixed range of levels and one queue per priority level, though it would probably just complicate things.
- tetron 11y ago
  This is basically the architecture of Arvados Crunch: http://arvados.org http://arvados.org
  - nikolay 11y ago
    I almost skipped your comment as I misread as Church. Looks like a great project!
- nemothekid 11y ago
  My gripe with using a relational database as a priority queue is tends not to scale well. (Similarly as if you tried to use a relational database as a queue). Once you get to the 10s of millions of messages per day (for example ~8M at priority X and 2M at priority Y), performance tends to go down the drain (due to contention), disk space bloats (due to vacumming/deletes). I've yet to come across a good task scheduling system that will take into the account resources available and the priority of the tasks, while still usable on a largish dataset that isn't a pain to maintain.
boomzilla 11y ago
What are the downsides with SQS if you expect a small number of large massages (well, not larger than the 4K limit I think)? The bill should be a lot cheaper than running a broker your self. Are you worried about lock-in? SQS APIs are simple enough to replicate with a REST service backed by a database for low usage scenarios.

by Jasper_ 11y ago

lobster_johnson 11y ago

I would argue that job management is unrelated to messaging, at least according my loose definition of a task that: * Has a well-defined lifetime — unstarted, running, paused, completed successfully, or failed; * Is executed from some kind of parameterized "job specification" that describes its inputs and desired behavior; * Has state data (e.g., completion % progress, log output, metrics, transactional continutation state for pausing/retrying); * Can be scheduled on multiple nodes; * Can be created and scheduled multiple times; * Can be scheduled by priority; * Can have its resources (RAM, CPU, I/O) constrained. The thing is, queues are terrible at this. People continue to abuse messaging systems as a scheduler, and it always sucks. For example, consider acking: For a long-running tasks that might take hours, do you really want to hold up the broker's internal lock for all that time? What if you restart the broker — the task is still running, now you decoupled its running state from the queue. Similarly, queues are terrible at finding out what's waiting, or what has happened. By using the queue as job state, you are coupling scheduling with execution, which is just wrong. A queue (one that supports priorities, mind you) is, however, excellent for management commands: "Start job X", "pause job X" and so on. But you will want to maintain execution state separately from the queue. A relational database is quite good at this, though one could use a NoSQL databases, too. First, create a row representing the job: {id="job1", state="new", type="import_stuff", params="..."}. Then push a queue message {commmand="start_job", job_id="job1"}. Run a queue consumer or ten on each node. Each consumer acks the message, finds the job in the table and starts it. One could run it as a child process, or in a Docker container. One could of course use an orchestration service (like Mesos/Marathon) to run it, and monitor the service for container events and update the job table appropriately. I would want to take it further and allow the job to update itself through API, so that it could publish metrics and log messages, as well as maintain whatever data is useful snapshotting. For example, a long-running task to import data into a database would benefit from maintaining a cursor so that you could pause/kill the job and later resume it at the point where it was.

Jasper_ 11y ago
Sure, that's actually very similar to the the system we built internally. But I don't see how Kafka as a messaging layer helps in your system, either. Presumably, the consumers won't advance their offset in their partition until the job is done, because they are not ready to start a new job. So you will still get the same blocking behavior. In addition, I don't see how you implement priority in a system like this, since Kafka partitions are append-only. One has to process the former start_job messages before you reach the higher priority items. Having a broker hand out items in priority to workers makes a lot more sense to me. The job system you described is definitely something missing from the OSS world, and we'd love to open up our system for this eventually. While Disque is not that piece, I think it makes more sense than Kafka.
- lobster_johnson 11y ago
  I didn't mean to imply that Kafka was suitable for such a system, though that's mostly because it doesn't support priority queues. You could work around this by using a fixed range of levels and one queue per priority level, though it would probably just complicate things.
tetron 11y ago
This is basically the architecture of Arvados Crunch: http://arvados.org http://arvados.org
- nikolay 11y ago
  I almost skipped your comment as I misread as Church. Looks like a great project!
nemothekid 11y ago
My gripe with using a relational database as a priority queue is tends not to scale well. (Similarly as if you tried to use a relational database as a queue). Once you get to the 10s of millions of messages per day (for example ~8M at priority X and 2M at priority Y), performance tends to go down the drain (due to contention), disk space bloats (due to vacumming/deletes). I've yet to come across a good task scheduling system that will take into the account resources available and the priority of the tasks, while still usable on a largish dataset that isn't a pain to maintain.

boomzilla 11y ago

What are the downsides with SQS if you expect a small number of large massages (well, not larger than the 4K limit I think)? The bill should be a lot cheaper than running a broker your self. Are you worried about lock-in? SQS APIs are simple enough to replicate with a REST service backed by a database for low usage scenarios.