When I started to write a lambda that issues a Redshift COPY I thought that by default the lambda would read the data it was copying, and that's not the case (even though it could be), so you could be copying duplicated data without ever noticing it, or burning retries on poisoned data. This post will try to explain the difference between control planes and data planes, and how an AWS COPY lambda that is triggered by the creation of SQS messages fits into it.
Control planes and data planes
To clarify, the term plane refers to a layer of the architecture (originally a network one, and only conceptually) where some specific kind of task is performed, and its meaning is defined by the boundary that you are drawing.
So the difference between control and data planes in general is the following:
Control plane: it's the decision maker, the one that decides where the data should go, and does the administrative work, the setup, the security policies.
Data plane: this one carries out those decisions, moves the data from source to destination, and follows the rules determined by the control plane. It's the one that actually touches the data.
AWS has a different concept, because for them control and data plane can be differentiated inside the same service. For example, Lambda's control plane is CreateFunction, and its data plane is Invoke. In my context, these concepts involve the ingestion pipeline and not the service itself. The boundary I am drawing is basically whose memory the bytes pass through. In other words, by AWS definition this lambda does call a data-plane API, redshift-data:ExecuteStatement, since Redshift is doing its primary job there and not a management call.
In the ingestion pipeline, the COPY lambda seems at the beginning to be a data plane component, but when we look closer at the implementation we notice that the data is not read at all, because the lambda only tells Redshift to move it. It hands over a pointer (the JSON manifests) and the bytes of the real data go to the Redshift cluster without even touching the lambda memory.
What the code actually does
To do a COPY operation that batch loads the data into a Redshift cluster we don't need to read the data at all, and reading it would only give the overall lambda more points of failure. Here is a simple implementation of it, with some error handling and configuration loading left out:
import json
import time
import boto3
s3 = boto3.client("s3")
redshift = boto3.client("redshift-data")
MAX_POLL_ATTEMPTS = 60
POLL_INTERVAL_SECONDS = 1
def handler(event, context):
urls = []
failures = []
for record in event["Records"]: # SQS messages announcing files, not the files
try:
message = json.loads(record["body"])
urls.extend(message["urls"])
except (json.JSONDecodeError, KeyError):
failures.append({"itemIdentifier": record["messageId"]})
manifest = {"entries": [{"url": url, "mandatory": True} for url in urls]}
key = f"manifests/{time.time_ns()}.json"
s3.put_object(Bucket=BUCKET, Key=key, Body=json.dumps(manifest))
statement = redshift.execute_statement(
ClusterIdentifier=CLUSTER,
Database=DATABASE,
Sql=f"COPY raw_events FROM 's3://{BUCKET}/{key}' IAM_ROLE '{IAM_ROLE}' FORMAT AS JSON 'auto' MANIFEST",
)
status = "SUBMITTED"
for _ in range(MAX_POLL_ATTEMPTS):
status = redshift.describe_statement(Id=statement["Id"])["Status"]
if status not in ("SUBMITTED", "PICKED", "STARTED"):
break
time.sleep(POLL_INTERVAL_SECONDS)
if status != "FINISHED": # one COPY, so the whole batch goes back to the queue
return {"batchItemFailures": [{"itemIdentifier": r["messageId"]} for r in event["Records"]]}
return {"batchItemFailures": failures}
If we look at the two clients created at the top we can see the whole point of this post, because this function only talks to S3 through put_object and to Redshift through execute_statement, and there is no get_object anywhere and no third client. The urls that come inside the message are copied into the manifest as strings and are never opened, so the lambda only writes down where the data is and asks Redshift to go and get it.
As we can see we produce a single manifest based on all the records of the SQS event, but if for some reason (like that one that I will explain in the next section) any of the messages is re-enqueued to SQS we could face duplication issues, since by nature the COPY statement is not idempotent. And since the COPY is done over a manifest, when the message comes back the whole manifest is copied again, so a retry duplicates a batch and not only one record.
What could break
The code above never validates the data. It validates the message that points to it, which is not the same thing. The records are basically references to the real data, so if there are some errors in there, the COPY will fail on Redshift's side and the lambda has no way of knowing about it beforehand. This means that the COPY statement is the first schema-enforcing read, and nothing checks those bytes against the destination schema until COPY does it.
Also, without a proper handling of it in a different lambda or a proper configuration of the DLQ (dead-letter queue), the same poison message could go back to the SQS queue again and again, resulting in the same error until the message simply expires. Let's imagine a scenario where we have a visibility timeout of 15 minutes and a redrive policy with a maxReceiveCount of 5. The visibility timeout has to be that large because it needs to cover the whole invocation, including the time the lambda can spend polling, otherwise SQS would hand the same message to a second invocation while the first one is still waiting for its COPY to finish.
With those numbers the retry story goes like this. The poison message is received, the COPY fails, and 15 minutes later the message becomes visible again, and this repeats until the fifth receive, when SQS finally moves it to the DLQ, where it will wait for the retention period (up to 14 days) until somebody looks at it. Until these limits are reached, the message will return to this lambda and we will replicate the same errors.
It is also worth saying that returning batchItemFailures is what makes this bearable. With ReportBatchItemFailures enabled on the event source mapping, only the messages listed in that return go back to the queue, and not the whole batch. The interesting part is that we can only be precise about half of it, because a message that is not valid JSON or that doesn't have the urls key is something the lambda has in memory and can check, so we are able to send back exactly those ids. In the COPY context, this is a different story, since there is a single statement over a single manifest, and the smallest thing that can fail there is the whole batch, so one bad file inside it sends every message back and on the retry every good file of that batch is loaded again. In other words, the lambda can be specific about the messages that it reads, and can only be generic about the data that it never touches.
The poll loop has a similar problem. When MAX_POLL_ATTEMPTS runs out we stop watching the statement, but stopping the watch doesn't stop the COPY, and it keeps running on the cluster. The messages go back to the queue, the next invocation writes a new manifest with the same urls, and we end up with two statements loading the same files. We could call cancel_statement in that branch, but this would only help while the COPY is still running, and by the time the poll gives up it could already have loaded everything, so the duplicate would happen anyway.
Another scenario is sending the message without thinking about idempotency. Redshift doesn't enforce primary keys or unique constraints, so if the same message is delivered twice (and with a standard SQS queue, at-least-once delivery means it can be), we could have two correct messages and duplicated entries on Redshift.
Possible fixes
In order to fix the errors described above we can develop some types of solution. For example, for avoiding poisoned messages, we should have some other lambda that will validate them before they reach SQS. Another option is to validate them on the same COPY lambda, which to me is not a good solution, since it would imply downloading the real data from the source, and it would drag the lambda onto the data path and detour it from its original goal.
For the idempotency issue, again we can have another lambda before this one to be responsible for creating some idempotency keys and passing this information to this lambda, or we can also do this work here, but again it will be another case of a component with multiple responsibilities (data plane and control plane at the same time), which is not the desired behavior. For this case the best solution is probably letting a Redshift statement be responsible for guaranteeing idempotence by discarding the duplicates. The query below, for example, can run as a model in an analytics tool like dbt, and everything downstream would read that model instead of reading the raw table:
with ranked as (
select
event_id,
payload,
loaded_at,
row_number() over (
partition by md5(payload)
order by loaded_at
) as duplicate_rank
from raw_events
)
select
event_id,
payload,
loaded_at
from ranked
where duplicate_rank = 1
The deduplication key here is the hash of the whole payload, which already carries the identifier of the event inside it, row_number() numbers the rows inside each group of identical rows, and keeping only the rank 1 collapses each group into a single row. It is important to notice that this one keeps one row per group, because the variant that filters by count(*) over (partition by ...) = 1 looks very similar but it throws the whole duplicated group away, losing the record instead of deduplicating it. The order by only decides which copy survives, and since the rows are identical it doesn't matter which one it is.
The most attentive readers have probably already noticed that the COPY itself is still not idempotent, but since we are thinking in the context of an overall implementation of the pipeline, the final goal is to guarantee that whatever reads the data downstream doesn't see duplicated rows. Since there is always the need to run the statement above we can say that this pipeline is kind of eventually idempotent, but the call of the lambda itself is not. The word eventually is important here, because the duplicated rows are still physically in the raw table, and anything that queries it before the deduplication runs will still see both copies. So the window where the data is wrong is the interval between the COPY and the next run of that model, which means that this window is not defined by the lambda but by the schedule that deduplication tool is running on. And anything that decides to read the raw table directly instead of the model is not covered by this at all.
There is also a second thing that this design does not guarantee. Since the key is a hash of the payload, two events that are really distinct but happen to be identical are indistinguishable from a redelivery, so the deduplication will drop one of them. I accepted this because in this pipeline an identical duplicate is always a retry, but this is an assumption about the data and not a property of the code, so if one day this stops being true the statement will silently discard real rows.
Conclusion
This COPY lambda was an example to me that even a simple lambda function responsible only for creating manifests and copying data can lead to wrong assumptions if you don't look into more details, and can also introduce a kind of error that a simple look into the happy path can hide. Also, the same lambda can handle the validation of malformed data or the deduplication, but this will contradict the single responsibility principle (becoming a control and data plane at the same time). The validation fits better in the function that writes the file, because that one already has the bytes in memory and checking them there doesn't cost anything extra, while doing it here would mean downloading the same data a second time only to look at it. So having a clear vision of each step of the function and a solution to each fault scenario can improve the chances of the overall implementation not becoming a complete failure.
