4 ms·

Another neat hack is to use Postgres as a quick & dirty replacement for Hadoop/MapReduce if you have a job that has big (100T+) input data but small (~1G) outpu

Another neat hack is to use Postgres as a quick & dirty replacement for Hadoop/MapReduce if you have a job that has big (100T+) input data but small (~1G) output data. A lot of common tasks fall into this category: generating aggregate statistics from large log files, searching Common Crawl for relevant webpages, identifying abusive users or transactions, etc. The architecture is to stick a list of your input shards in a Postgres table, have a state flag that goes PENDING->WORKING->FINISHED->(ERROR?), and then spin up a bunch of worker processes as EC2 spot instances that check for the next PENDING task, mark it as WORKING, pull it, process it, mark it as FINISHED, and repeat. They write their output back to the DB in a transaction; there's an assumption that aggregation can happen in-process and then get merged in a relatively cheap transaction. If the worker fails or gets pre-empted, it retries (or marks as ERROR) any shards it was previously working on. Postgres basically functions as the MapReduce Master & Reducer, the worker functions as the Mapper and Combiner, and there's no need for a shuffle phase because output <<< input. Almost all the actual complexity in MapReduce/Hadoop is in the shuffle, so if you don't need that, the remaining stuff takes < 1 hour to implement and can be done without any frameworks.

ones_and_zeros 7y ago
If a worker fails or gets pre-empted, how does it retry anything? It's gone at that point, no? Sounds like you'd end up with a bunch of dangling shards orphaned in WORKING state. And now you need timeouts and health checks and something to coordinate all that.
- nostrademons 7y ago
  You've got your normal AWS orchestration working on top of this. If a spot instance is pre-empted AWS will automatically re-start it once the current spot price falls below your bid. One of the parameters you pass to the spot instance is a shell script to run when the instance starts, so that shell script just launches your worker process and in your worker process's main() it checks for the existence of previous WORKING shards with the same machine ID and marks them back as PENDING. You can (and probably want to) also have a retry count that gets incremented so that if there's a problem with the data that results in a permanent failure, you give up, mark it as ERROR (logging the offending shard, record, and error message if possible), and move on to the next shard. For exceptions, crashes, and other unexpected errors, you just let it crash. In the shell script that starts the worker process, you wrap it in "while (1) { ... }", so it's just permanently restarting. Then the aforementioned retry check in main() handles the previous shard just as if the worker had been pre-empted or finished normally. There are a variety of low-tech ways to detect when the job as a whole has finished, ranging from manually shutting down the cluster to writing a small cronjob that counts PENDING/WORKING shards and shuts down the cluster if there are none to having your workers return a different exit status if there's no more work to do and shutting down the box if so. A neat side effect of this is that you get an easy status report of exactly which machine is working on which shard and how long you have to go through "SELECT * FROM work_queue". Another neat side effect is that you can view partial results in the DB before the job as a whole completes, something you can't do with MapReduce, and potentially adjust your code and restart the job if you're getting bad data.
- foota 7y ago
  Cloud things provide retry of tasks, so you can kind of assume they'll get done eventually. I haven't seen a system relying on this though.
- ris 7y ago
  I used to do something a little like this with postgres advisory locks. Worker takes out a lock that signifies it's working on a task, if the worker fails or loses its connection in any way the lock gets dropped along with the transaction being rolled back. Example: https://gist.github.com/risicle/f4807bd706c9862f69aa https://gist.github.com/risicle/f4807bd706c9862f69aa
  - rem7 7y ago
    Yeah, in MySQL you can also do this with a named lock. Really like this workflow.
- lallysingh 7y ago
  SQL is pretty powerful. A last_updated column and a PENDING view that shows PENDING and WORKING with now - last_updated > CUTOFF does a decent job for you.
nurettin 7y ago
>> spin up a bunch of worker processes as EC2 spot instances that check for the next PENDING task Does that mean workers poll the database?
- nostrademons 7y ago
  Yup, but they do so only once per shard (well, twice, once at the beginning and once at the end). If you've got a big job where each shard takes a few minutes to process and a hundred or so workers, the DB gets about 1 req/sec, which is well within the capabilities of Postgres.
  - flavor8 7y ago
    We had an "interesting" bug where we inadvertently ended up polling postgres advisory locks around 6 million times per minute from a cluster of 200 servers. It added about 20% load to the database, enough to be anomalous but not enough to outright trip any alarms. The overall database performance was fine.
    - exikyut 7y ago
      (That's 100,000 req/sec) Was "the database" a single node, or multiple systems?
  - nurettin 7y ago
    I would lean towards maintaining a simple list on redis and letting the workers pop the next job from it whenever they are done.
    - nostrademons 7y ago
      That works fine too. The advantage of Postgres is that the master is a single point of failure, and if Postgres goes down all the state is persistent and you can just bring it back up again. Also, ad-hoc query tools are a little better for Postgres, which gives you a really simple interface to inspect the status of the job and the partial results. This architecture originally evolved from an Amazon-SQS based queue system, where I'd stick the jobs in SQS and pop them off. Then I realized that I needed the DB anyway for my reducer-substitute, and if I'm writing SQL/ORM code anyway it was simpler to use the DB to manage the queue than to use SQS, and it gave me some other nifty abilities like being able to record timing & size stats with each shard, being able to query progress while the job is going, and being able to manually restart shards after correcting the errors that led to their failure. If you're running dozens-to-hundreds of workers, the cost of a t2 RDS instance (that's all you need for 1 QPS) is a rounding error, and a lot cheaper than EMR would be.
ithkuil 7y ago
It reminds me of the taskmaster tool my team used at Google (https://landing.google.com/sre/sre-book/chapters/data-processing-pipelines/ https://landing.google.com/sre/sre-book/chapters/data-proces...)
- ithkuil 7y ago
  Care to explain the downvotes?
takeda 7y ago
as a person that had to maintain a hadoop cluster, well it's def not quick, but I can't imagine postgres being dirtier.
rhizome 7y ago
All the replies to this and no mention of the workflow pattern! I guess Google went ahead and NIH'ed it too, to boot! Surely some of those people had programmed a workflow before, somewhere along the line. I'm not going to say "I don't know why anybody is surprised," but I'll tell you that I'm surprised! This is good, you're all one of 10,000![1] I'd call parent's a "basic" workflow, but Wikipedia seems to tag it as "state-based."[2] I'm glad someone pointed out that failed steps do not update the db unless you have good traps in the process itself. I'd probably suggest a housekeeping process that resets any errors after some amount of time so that they're run through again from the beginning. As long as the error is surfaced in the meantime, that is. 1. https://xkcd.com/1053/ https://xkcd.com/1053/ 2. https://en.wikipedia.org/wiki/Workflow_pattern https://en.wikipedia.org/wiki/Workflow_pattern
- nostrademons 7y ago
  This was actually for a few of my post-Google projects. At Google we would just use MapReduce regardless of how inappropriate it was because of the difficulty of standing up an RDBMS on Google's cloud infrastructure, plus CPU cycles were basically free for engineers. Maybe it's different now that internal projects are encouraged to use Google Cloud; Cloud SQL + Compute Engine works just fine for this. And this is definitely not a new pattern - if you wanna go back in time, this is how things often worked in the mainframe era. My point - and IIUIC the point of the article - is that people often reach for flavor-of-the-month heavyweight frameworks when they can easily adapt tools they're already using to solve the problem within hours. (Another great example of this is the single-process in-RAM architecture used by PlentyOfFish, StackOverflow, Mailinator, and Hacker News when they were solo-developer projects, where you store all state in RAM and run the whole site on a single box, periodically dumping out checkpoints to disk in case the process crashes. This can scale to thousands of requests per second - that's millions of queries per day - and is a lot simpler to program than 3-tier architectures where everything runs through a data abstraction layer and you spend 90+% of the code shuffling data around.)
  - gspetr 7y ago
    > CPU cycles were basically free for engineers. So what happened to a guy who had the bright idea to do CPU mining on Google's infra?
dplgk 7y ago
Forgive my ignorance, how is this specific to postgres? Or what are the advantages of doing this in pg rather than mysql or redis?
- nostrademons 7y ago
  It's not really, although you really want to do it with a data store that supports transactional semantics. Postgres, MySQL with InnoDB,Redis, Amazon SQS, DynamoDB will all work. First time I used this pattern was with Amazon SQS; second time was MySQL with InnoDB tables. The article's about Postgres hacks, though, and Postgres is pretty commonly used for other parts of your system.
BukhariH 7y ago
That's a very similar pattern to what I've been using at TW. At the core of our implementation is this library to schedule the jobs to different workers: https://github.com/kagkarlsson/db-scheduler https://github.com/kagkarlsson/db-scheduler Would highly recommend that library!

ones_and_zeros 7y ago

If a worker fails or gets pre-empted, how does it retry anything? It's gone at that point, no? Sounds like you'd end up with a bunch of dangling shards orphaned in WORKING state. And now you need timeouts and health checks and something to coordinate all that.

nostrademons 7y ago
You've got your normal AWS orchestration working on top of this. If a spot instance is pre-empted AWS will automatically re-start it once the current spot price falls below your bid. One of the parameters you pass to the spot instance is a shell script to run when the instance starts, so that shell script just launches your worker process and in your worker process's main() it checks for the existence of previous WORKING shards with the same machine ID and marks them back as PENDING. You can (and probably want to) also have a retry count that gets incremented so that if there's a problem with the data that results in a permanent failure, you give up, mark it as ERROR (logging the offending shard, record, and error message if possible), and move on to the next shard. For exceptions, crashes, and other unexpected errors, you just let it crash. In the shell script that starts the worker process, you wrap it in "while (1) { ... }", so it's just permanently restarting. Then the aforementioned retry check in main() handles the previous shard just as if the worker had been pre-empted or finished normally. There are a variety of low-tech ways to detect when the job as a whole has finished, ranging from manually shutting down the cluster to writing a small cronjob that counts PENDING/WORKING shards and shuts down the cluster if there are none to having your workers return a different exit status if there's no more work to do and shutting down the box if so. A neat side effect of this is that you get an easy status report of exactly which machine is working on which shard and how long you have to go through "SELECT * FROM work_queue". Another neat side effect is that you can view partial results in the DB before the job as a whole completes, something you can't do with MapReduce, and potentially adjust your code and restart the job if you're getting bad data.
foota 7y ago
Cloud things provide retry of tasks, so you can kind of assume they'll get done eventually. I haven't seen a system relying on this though.
ris 7y ago
I used to do something a little like this with postgres advisory locks. Worker takes out a lock that signifies it's working on a task, if the worker fails or loses its connection in any way the lock gets dropped along with the transaction being rolled back. Example: https://gist.github.com/risicle/f4807bd706c9862f69aa https://gist.github.com/risicle/f4807bd706c9862f69aa
- rem7 7y ago
  Yeah, in MySQL you can also do this with a named lock. Really like this workflow.
lallysingh 7y ago
SQL is pretty powerful. A last_updated column and a PENDING view that shows PENDING and WORKING with now - last_updated > CUTOFF does a decent job for you.

nurettin 7y ago

>> spin up a bunch of worker processes as EC2 spot instances that check for the next PENDING task Does that mean workers poll the database?

nostrademons 7y ago
Yup, but they do so only once per shard (well, twice, once at the beginning and once at the end). If you've got a big job where each shard takes a few minutes to process and a hundred or so workers, the DB gets about 1 req/sec, which is well within the capabilities of Postgres.
- flavor8 7y ago
  We had an "interesting" bug where we inadvertently ended up polling postgres advisory locks around 6 million times per minute from a cluster of 200 servers. It added about 20% load to the database, enough to be anomalous but not enough to outright trip any alarms. The overall database performance was fine.
  - exikyut 7y ago
    (That's 100,000 req/sec) Was "the database" a single node, or multiple systems?
- nurettin 7y ago
  I would lean towards maintaining a simple list on redis and letting the workers pop the next job from it whenever they are done.
  - nostrademons 7y ago
    That works fine too. The advantage of Postgres is that the master is a single point of failure, and if Postgres goes down all the state is persistent and you can just bring it back up again. Also, ad-hoc query tools are a little better for Postgres, which gives you a really simple interface to inspect the status of the job and the partial results. This architecture originally evolved from an Amazon-SQS based queue system, where I'd stick the jobs in SQS and pop them off. Then I realized that I needed the DB anyway for my reducer-substitute, and if I'm writing SQL/ORM code anyway it was simpler to use the DB to manage the queue than to use SQS, and it gave me some other nifty abilities like being able to record timing & size stats with each shard, being able to query progress while the job is going, and being able to manually restart shards after correcting the errors that led to their failure. If you're running dozens-to-hundreds of workers, the cost of a t2 RDS instance (that's all you need for 1 QPS) is a rounding error, and a lot cheaper than EMR would be.

ithkuil 7y ago

It reminds me of the taskmaster tool my team used at Google (https://landing.google.com/sre/sre-book/chapters/data-processing-pipelines/ https://landing.google.com/sre/sre-book/chapters/data-proces...)

ithkuil 7y ago
Care to explain the downvotes?

takeda 7y ago

as a person that had to maintain a hadoop cluster, well it's def not quick, but I can't imagine postgres being dirtier.

rhizome 7y ago

All the replies to this and no mention of the workflow pattern! I guess Google went ahead and NIH'ed it too, to boot! Surely some of those people had programmed a workflow before, somewhere along the line. I'm not going to say "I don't know why anybody is surprised," but I'll tell you that I'm surprised! This is good, you're all one of 10,000![1] I'd call parent's a "basic" workflow, but Wikipedia seems to tag it as "state-based."[2] I'm glad someone pointed out that failed steps do not update the db unless you have good traps in the process itself. I'd probably suggest a housekeeping process that resets any errors after some amount of time so that they're run through again from the beginning. As long as the error is surfaced in the meantime, that is. 1. https://xkcd.com/1053/ https://xkcd.com/1053/ 2. https://en.wikipedia.org/wiki/Workflow_pattern https://en.wikipedia.org/wiki/Workflow_pattern

nostrademons 7y ago
This was actually for a few of my post-Google projects. At Google we would just use MapReduce regardless of how inappropriate it was because of the difficulty of standing up an RDBMS on Google's cloud infrastructure, plus CPU cycles were basically free for engineers. Maybe it's different now that internal projects are encouraged to use Google Cloud; Cloud SQL + Compute Engine works just fine for this. And this is definitely not a new pattern - if you wanna go back in time, this is how things often worked in the mainframe era. My point - and IIUIC the point of the article - is that people often reach for flavor-of-the-month heavyweight frameworks when they can easily adapt tools they're already using to solve the problem within hours. (Another great example of this is the single-process in-RAM architecture used by PlentyOfFish, StackOverflow, Mailinator, and Hacker News when they were solo-developer projects, where you store all state in RAM and run the whole site on a single box, periodically dumping out checkpoints to disk in case the process crashes. This can scale to thousands of requests per second - that's millions of queries per day - and is a lot simpler to program than 3-tier architectures where everything runs through a data abstraction layer and you spend 90+% of the code shuffling data around.)
- gspetr 7y ago
  > CPU cycles were basically free for engineers. So what happened to a guy who had the bright idea to do CPU mining on Google's infra?

dplgk 7y ago

Forgive my ignorance, how is this specific to postgres? Or what are the advantages of doing this in pg rather than mysql or redis?

nostrademons 7y ago
It's not really, although you really want to do it with a data store that supports transactional semantics. Postgres, MySQL with InnoDB,Redis, Amazon SQS, DynamoDB will all work. First time I used this pattern was with Amazon SQS; second time was MySQL with InnoDB tables. The article's about Postgres hacks, though, and Postgres is pretty commonly used for other parts of your system.

BukhariH 7y ago

That's a very similar pattern to what I've been using at TW. At the core of our implementation is this library to schedule the jobs to different workers: https://github.com/kagkarlsson/db-scheduler https://github.com/kagkarlsson/db-scheduler Would highly recommend that library!