One order file, three services
A cloud pipeline is a chain of services that passes work along, one step to the next, with nobody pressing a button. Most real cloud systems are built this way, so you need to know what happens when a step fails or runs twice.
A customer uploads an order file to an S3 bucket. A minute later the order is a row in a DynamoDB table. Three small steps did it: an event, a queue and a function.
By the end of this lesson
Note
I can follow one upload through a queue and a function to a table.
Key words
| Word | What it means | Example |
|---|---|---|
| Pipeline | steps that pass work along, one to the next | |
| Event | a note that something happened | |
| Message | one piece of text waiting in a queue | |
| Item | one row stored in a DynamoDB table | |
| Idempotent | safe to run twice, with the same result | put an item only if it is new |
The cells students fill in are left blank in this preview.
Each step hands text to the next
The upload makes S3 send an event. The bucket is set up to send that event to an SQS queue as a message. A function reads the message and writes an item to a table.
The queue carries text, not dictionaries. So the event is turned into a JSON string, and the function turns it back. This event is the documented shape, not one a stack made for you.
Run it: an event becomes a message body
import json
# The documented shape of an S3 event, cut down to what we read.
event = {"Records": [{
"eventName": "ObjectCreated:Put",
"s3": {"bucket": {"name": "shop-uploads"},
"object": {"key": "orders/ana/3.json"}}}]}
# The queue carries text, so the event becomes a JSON string.
message = {"MessageId": "m-1", "ReceiptHandle": "h-1", "Body": json.dumps(event)}
print(type(message["Body"]).__name__)
# The worker turns the text back into a dictionary.
record = json.loads(message["Body"])["Records"][0]
print(record["s3"]["object"]["key"])
parts = record["s3"]["object"]["key"].split("/")
print(parts)
# Change the key to "orders/ben/7.json" and run again.
A message can arrive twice
A queue promises at least once. Now and then it hands over the same message again, for example when a worker was slow to finish.
So the function must be idempotent. Writing the order again must not make a second order. A conditional write does this: put the item only if nothing is stored under that key.
A common mistake
Warning
Many people think each message arrives exactly once. It does not: a queue may deliver one more than once. Write every handler so a repeat changes nothing. The second write is not an error to crash on, it means the work is done.
Run it: the second write is refused
table = {}
def put_item_once(table, item):
key = item["customer"]["S"] + "#" + item["order_id"]["N"]
if key in table:
return {"error": {"Code": "ConditionalCheckFailedException",
"HTTPStatusCode": 400}}
table[key] = item
return {}
item = {"customer": {"S": "ana"}, "order_id": {"N": "3"},
"status": {"S": "received"}}
again = item
first = put_item_once(table, item)
second = put_item_once(table, again)
print(first)
print(second)
print(len(table))
# Change again = item to again = {"customer": {"S": "ana"}, "order_id": {"N": "4"}}.
# How many items are stored now?
Delete a message after its work is done
A worker takes a batch of messages. One of them is broken. The others are fine.
Handle each message on its own. After the work is saved, delete that message. A message you could not handle stays in the queue and comes back after its visibility timeout. Nothing is lost, and nothing is deleted too early.
Run it: a batch with one bad message
queue = [
{"MessageId": "m-1", "Body": "orders/ana/3.json"},
{"MessageId": "m-2", "Body": "broken"},
{"MessageId": "m-3", "Body": "orders/ben/7.json"},
]
done = []
waiting = []
for message in queue:
parts = message["Body"].split("/")
if len(parts) == 3:
done.append(message["MessageId"])
else:
waiting.append(message["MessageId"])
print("deleted:", done)
print("left:", waiting)
# Change "broken" to "orders/cai/9.json". What is left now?