In this post we are going to cover loading and moving data in Azure Data Lake Storage Gen2: picking a transfer tool, AzCopy, storage events with Event Grid, the change feed, copies between accounts and tracking processed files. These are 7 real ADLS Gen2 and Blob Storage interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: hundreds of partner files arrive every day, and each must be validated and loaded within ten minutes.
Moving data is most of the daily work around a lake. Files come in from on-premises, from partners and from other accounts, and every one of them has to be loaded once, on time, and only when it is complete. Interviewers ask about this because the tool choice is easy to get wrong, and because the quiet bugs live here: half-written files, missed files and files loaded twice.
Part 7 of the Real Time Azure Data Lake Storage Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A teammate must copy ten terabytes from on-premises into the lake. Which tool do you use?
- Question 2: Someone asks how AzCopy works and when to use it. What do you tell them?
- Question 3: A pipeline must start as soon as a file lands. How do storage events and Event Grid help?
- Question 4: A teammate asks how to find out which blobs changed since yesterday. How does the blob change feed work?
- Question 5: Someone asks how to copy data between two storage accounts. What are the options?
- Question 6: A pipeline reprocesses files it already loaded. How do you track what has been processed?
- Scenario 7: Hundreds of partner files arrive every day and each must be validated and loaded within ten minutes. How do you build it?
- The full script
The short answer
Do the transfer math first: AzCopy for bulk copies, Data Factory for scheduled pipelines, Data Box for slow links. Start loads from Event Grid with an API filter for complete files; the change feed isn't available with a hierarchical namespace. Copy between accounts server side, and track processed files by name and etag.
Let's set up the project
We run Azurite, Microsoft's open-source Azure Storage emulator, locally, and small Python scripts with the azure-storage-blob client, so every output below is real. Features the emulator does not support, like Event Grid, the change feed and AzCopy against Azure, are shown as code. common.py creates the client from the connection string and has a small container() helper that creates a container if it is missing.
import os
from azure.core.exceptions import ResourceExistsError
from azure.storage.blob import BlobServiceClient
svc = BlobServiceClient.from_connection_string(
os.environ["AZURE_STORAGE_CONNECTION_STRING"])
def container(name, client=svc):
"""Create the container if it is missing, return its client."""
try:
return client.create_container(name)
except ResourceExistsError:
return client.get_container_client(name)Question 1: A teammate must copy ten terabytes from on-premises into the lake. Which tool do you use?
I do the math before I pick the tool. A decent link means AzCopy, a scheduled pipeline means Data Factory, and a slow or busy link with lots of data means Data Box.
Ten terabytes is 80 trillion bits. transfer_time.py divides that by the link speed, at full line rate. We run python transfer_time.py.
# How long does 10 TB take over the network at full line rate?
SIZE_TB = 10
bits = SIZE_TB * 10**12 * 8 # 1 TB = 10**12 bytes, not TiB
for link, bps in [("100 Mbps", 100e6), ("1 Gbps", 1e9),
("10 Gbps", 10e9)]:
hours = bits / bps / 3600
print(f"{SIZE_TB} TB over {link:>8}: {hours:6.1f} hours"
f" = {hours / 24:4.1f} days")$ python transfer_time.py
10 TB over 100 Mbps: 222.2 hours = 9.3 days
10 TB over 1 Gbps: 22.2 hours = 0.9 days
10 TB over 10 Gbps: 2.2 hours = 0.1 daysAt 1 Gbps, that is 22.2 hours. At 100 Mbps, 222.2 hours, more than nine days. At 10 Gbps, about 2.2 hours. Real throughput is lower, so I add margin.
| Tool | Use it when | Notes |
|---|---|---|
| AzCopy v10 | the network can move it in time | parallel, resumable; v10 for HNS accounts |
| Data Factory | the copy becomes a scheduled pipeline | self-hosted integration runtime on-premises |
| Data Box (Disk) | slow or busy link; Data Box: over 40 TB | Data Box: 120 or 525 TB usable; Disk: 8 TB disks |
With a decent link, I use AzCopy over the network: it runs in parallel and can resume, and hierarchical namespace accounts need version 10. Data Factory fits when the copy becomes a scheduled pipeline, with a self-hosted integration runtime on-premises. Data Box is for a slow or busy link. Microsoft positions it for more than 40 TB with little network connectivity. For ten terabytes on a slow link, Data Box Disk ships 8 TB disks.
That link is shared with the business. Cap the bandwidth with azcopy --cap-mbps, run at night, and copy one folder first to measure the real speed.
Question 2: Someone asks how AzCopy works and when to use it. What do you tell them?
AzCopy is Microsoft's command-line tool for bulk copies to and from Blob Storage and ADLS. It splits files into blocks, moves them in parallel, and each run is a job with a log, so a failed job can be resumed.
Let's go through it in order. First, sign in with Microsoft Entra ID using azcopy login, or put a SAS on the URL. No account keys in scripts. Then, for copy, the flags that matter:
--recursive- copy is not recursive by default--put-md5- stores a hash on each blob--overwrite ifSourceNewer- only newer files replace yours
Here is a typical load script:
# 1. sign in with Microsoft Entra ID: no keys in the script
azcopy login
# 2. upload a folder: check with --dry-run, then run for real
azcopy copy "/data/sales" \
"https://lakeprod.blob.core.windows.net/raw" \
--recursive --include-pattern "*.csv" \
--put-md5 --overwrite ifSourceNewer --dry-run
# 3. a job failed halfway? resume it
azcopy jobs resume <job-id>
# 4. sync only new and changed files (recursive by default)
azcopy sync "/data/sales" \
"https://lakeprod.blob.core.windows.net/raw/sales" \
--delete-destination falsesync is different. It compares names and last modified times, and it is recursive by default. --delete-destination is false by default, and true deletes without a prompt, so turn on soft delete first.
| copy | sync | |
|---|---|---|
| compares | not by default | name + last modified |
| recursive | off by default | on by default |
| deletes | never | only --delete-destination true, no prompt |
I use AzCopy for one-off migrations, backfills and copies between accounts. For a scheduled load with retries and monitoring, I use Data Factory instead.
Two more tips:
--dry-run- see what would happen first--include-pattern- file names only, not paths
Run with --dry-run before every big copy or sync. It lists what would be copied or deleted, and it catches a filter that doesn't match what you meant.
Question 3: A pipeline must start as soon as a file lands. How do storage events and Event Grid help?
The storage account publishes events like BlobCreated and BlobDeleted to Event Grid. A filtered subscription pushes each event to a function, a queue or a webhook in near real time, so nothing polls the lake.
The flow looks like this:
- Storage account - BlobCreated
- Event Grid - filters
- Storage queue - buffers bursts
- Loader - function or job
The filters matter. A subject prefix limits it to the landing container. On a hierarchical namespace account, BlobCreated also fires on CreateFile, for a new, still empty file. So filter data.api to PutBlob, PutBlockList, CopyBlob or FlushWithClose.
LAKE_ID=$(az storage account show -n lakeprod -g rg-data \
--query id -o tsv)
az eventgrid event-subscription create --name landing-files \
--source-resource-id $LAKE_ID \
--endpoint-type storagequeue \
--endpoint $LAKE_ID/queueservices/default/queues/landing \
--included-event-types Microsoft.Storage.BlobCreated \
--subject-begins-with /blobServices/default/containers/landing/ \
--advanced-filter data.api StringIn PutBlob PutBlockList \
CopyBlob FlushWithClose \
--deadletter-endpoint \
$LAKE_ID/blobServices/default/containers/deadletterThe emulator has no Event Grid, so land.py uploads a partner file, and we hand handler.py two sample events in the documented shape. We run python land.py, then python handler.py event_create.json and python handler.py event_flush.json.
from common import container
csv = "order_id,amount\n101,20.5\n102,7.25\n103,12\n"
container("landing").upload_blob("acme/orders.csv", csv,
overwrite=True)
print("partner upload: landing/acme/orders.csv,", len(csv), "bytes"){
"eventType": "Microsoft.Storage.BlobCreated",
"subject":
"/blobServices/default/containers/landing/blobs/acme/orders.csv",
"data": {
"api": "CreateFile",
"contentLength": 0,
"blobType": "BlockBlob"
}
}{
"eventType": "Microsoft.Storage.BlobCreated",
"subject":
"/blobServices/default/containers/landing/blobs/acme/orders.csv",
"data": {
"api": "FlushWithClose",
"contentLength": 41,
"blobType": "BlockBlob"
}
}import json, sys
from common import svc
READY = {"PutBlob", "PutBlockList", "CopyBlob", "FlushWithClose"}
event = json.load(open(sys.argv[1]))
api = event["data"]["api"]
if api not in READY: # CreateFile: still empty
print(f"skip {api}: {event['data']['contentLength']} bytes")
sys.exit(0)
path = event["subject"].split("/containers/", 1)[1]
container, name = path.split("/blobs/", 1)
blob = svc.get_blob_client(container, name)
lines = blob.download_blob(encoding="utf-8").readall().splitlines()
print(f"load {container}/{name}: {len(lines) - 1} rows")$ python land.py
partner upload: landing/acme/orders.csv, 41 bytes
$ python handler.py event_create.json
skip CreateFile: 0 bytes
$ python handler.py event_flush.json
load landing/acme/orders.csv: 3 rowsThe CreateFile event, 0 bytes, is skipped. The FlushWithClose event loads landing/acme/orders.csv: 3 rows.
Events might arrive out of order, so make loads idempotent, and set a dead-letter container for events that can't be delivered.
Events go to a storage queue first, and the loader reads from it. The queue absorbs bursts, and retries are easy.
Question 4: A teammate asks how to find out which blobs changed since yesterday. How does the blob change feed work?
The change feed is an ordered, durable, read-only log of every create, update and delete, stored as Avro files in the $blobchangefeed container. Records appear within a few minutes.
It is ordered per blob, it keeps history as long as you set, and you can replay it, which Event Grid can't. The catch: it isn't supported with a hierarchical namespace, so an ADLS Gen2 lake has no change feed.
# Flat-namespace accounts only: not with a hierarchical namespace
az storage account blob-service-properties update \
--account-name stshopblob --resource-group rg-data \
--enable-change-feed true
# records: Avro files in the $blobchangefeed container,
# ordered per blob, published within a few minutes| Change feed | Event Grid | |
|---|---|---|
| delivery | a log you read, within minutes | pushed, near real time |
| order | per blob | might not be ordered |
| replay | yes, kept as long as you set | no log: retries, then dead-letter |
| hierarchical namespace | not supported | supported |
There I use stored Event Grid events, blob inventory reports, or for a small folder, a listing diff. Let's try the diff. seed_raw.py writes three files, snapshot.py saves each blob's name and etag, and todays_work.py corrects one file, adds one and deletes one. Then we take today's snapshot and compare. We run python seed_raw.py, python snapshot.py yesterday.json, python todays_work.py, python snapshot.py today.json, then python diff.py.
from common import container
raw = container("raw")
for day in ["2026-10-26", "2026-10-27", "2026-10-28"]:
raw.upload_blob(f"sales/{day}.csv", f"day,amount\n{day},100\n",
overwrite=True)
print("raw/sales: 3 files")import json, sys
from common import container
raw = container("raw")
state = {b.name: b.etag # name -> etag
for b in raw.list_blobs(name_starts_with="sales/")}
json.dump(state, open(sys.argv[1], "w"))
print("snapshot", sys.argv[1] + ":", len(state), "blobs")from common import container
raw = container("raw")
raw.upload_blob("sales/2026-10-28.csv",
"day,amount\n2026-10-28,120\n", overwrite=True) # fix
raw.upload_blob("sales/2026-10-29.csv", "day,amount\n2026-10-29,90\n")
raw.delete_blob("sales/2026-10-26.csv") # cleaned up
print("updated 1, added 1, deleted 1")import json
old = json.load(open("yesterday.json"))
new = json.load(open("today.json"))
print("new :", sorted(new.keys() - old.keys()))
print("changed:", sorted(k for k in new.keys() & old.keys()
if new[k] != old[k])) # etag moved
print("deleted:", sorted(old.keys() - new.keys()))$ python seed_raw.py
raw/sales: 3 files
$ python snapshot.py yesterday.json
snapshot yesterday.json: 3 blobs
$ python todays_work.py
updated 1, added 1, deleted 1
$ python snapshot.py today.json
snapshot today.json: 3 blobs
$ python diff.py
new : ['sales/2026-10-29.csv']
changed: ['sales/2026-10-28.csv']
deleted: ['sales/2026-10-26.csv']The diff shows exactly that: sales/2026-10-29.csv is new, sales/2026-10-28.csv changed because its etag moved, and sales/2026-10-26.csv is deleted. Blob inventory does the same at scale, with a daily or weekly report in CSV or Parquet.
Listing millions of blobs every day is slow, and it is billed. At that scale, diff two inventory reports instead.
Question 5: Someone asks how to copy data between two storage accounts. What are the options?
For bulk or one-off copies, AzCopy, server to server. For scheduled copies, a Data Factory copy activity with a managed identity. For a few files from code, Copy Blob. Object replication is out: it doesn't support a hierarchical namespace.
| Option | Best for | Note |
|---|---|---|
| AzCopy, server to server | bulk and one-off copies | Entra ID or SAS; data flows service to service |
| Data Factory copy activity | scheduled copies | managed identity, retries, monitoring |
| Copy Blob from code | a few files | start_copy_from_url with a source SAS |
| Object replication | not for ADLS Gen2 | not supported with a hierarchical namespace |
Let's do a Copy Blob. copy_accounts.py copies a file from the partnerlake account into our raw zone. The destination service reads the source itself, so the bytes never pass through our machine. It tries first without a SAS on the source URL, then with a one-hour read SAS. We run python copy_accounts.py.
import os, time
from datetime import datetime, timedelta, timezone
from azure.core.exceptions import HttpResponseError
from azure.storage.blob import (BlobServiceClient, BlobSasPermissions,
generate_blob_sas)
from common import container
partner = BlobServiceClient.from_connection_string(
os.environ["PARTNER_STORAGE_CONNECTION_STRING"])
src = container("exports", partner).upload_blob(
"orders/2026-10-29.csv", "order_id,amount\n1,20.5\n2,7.25\n")
dst = container("raw").get_blob_client("partner/2026-10-29.csv")
try:
dst.start_copy_from_url(src.url) # no SAS
except HttpResponseError as e:
print("copy without SAS:", e.error_code.value)
sas = generate_blob_sas(
partner.account_name, "exports", src.blob_name,
account_key=partner.credential.account_key,
permission=BlobSasPermissions(read=True),
expiry=datetime.now(timezone.utc) + timedelta(hours=1))
dst.start_copy_from_url(f"{src.url}?{sas}") # server-side copy
while dst.get_blob_properties().copy.status == "pending":
time.sleep(5) # Azure: can be async
props = dst.get_blob_properties()
print("copy with SAS:", props.copy.status, props.size, "bytes")$ python copy_accounts.py
copy without SAS: CannotVerifyCopySource
copy with SAS: success 30 bytesWithout a SAS, the copy fails with CannotVerifyCopySource. With the SAS, it works: success, 30 bytes. The emulator completes copies at once. On Azure, a copy between accounts can stay pending, so the script polls copy.status.
For AzCopy and Data Factory, prefer Microsoft Entra ID over a SAS. And keep both accounts in one region when you can, to avoid transfer charges between regions.
After a big copy, verify counts and sizes on both sides before anyone deletes the source.
Question 6: A pipeline reprocesses files it already loaded. How do you track what has been processed?
Keep a processed-files log, and make the load itself idempotent. The log records each file's name and version, and the loader skips what it has seen.
Here the pieces are:
landing/orders/- new and re-sent files- Loader - skip if name + etag seen
control/-orders_loaded.json
drop_files.py plays the partner, and load.py reads the log, loads anything new or changed, and writes the log back. We run the partner drop for the 28th and 29th, then python load.py twice. Then the partner sends the 29th again plus the 30th, and we load once more.
import sys
from common import container
landing = container("landing")
for day in sys.argv[1:]: # the partner (re)sends
landing.upload_blob(f"orders/{day}.csv",
f"order_id,day\n1,{day}\n", overwrite=True)
print("partner sent:", ", ".join(sys.argv[1:]))import json
from azure.core.exceptions import ResourceNotFoundError
from common import container
landing, control = container("landing"), container("control")
log = control.get_blob_client("orders_loaded.json")
try:
done = json.loads(log.download_blob().readall())
except ResourceNotFoundError:
done = {} # first run: empty log
loaded, skipped = [], 0
for b in landing.list_blobs(name_starts_with="orders/"):
if done.get(b.name) == b.etag: # same file, same version
skipped += 1
continue
# ... load b into curated, overwriting its own output ...
done[b.name] = b.etag
loaded.append(b.name)
log.upload_blob(json.dumps(done), overwrite=True)
print("loaded:", loaded, "| skipped:", skipped)$ python drop_files.py 2026-10-28 2026-10-29
partner sent: 2026-10-28, 2026-10-29
$ python load.py
loaded: ['orders/2026-10-28.csv', 'orders/2026-10-29.csv'] | skipped: 0
$ python load.py
loaded: [] | skipped: 2
$ python drop_files.py 2026-10-29 2026-10-30
partner sent: 2026-10-29, 2026-10-30
$ python load.py
loaded: ['orders/2026-10-29.csv', 'orders/2026-10-30.csv'] | skipped: 1The first run loads both files. The second loads nothing: skipped: 2. After the second drop, the re-sent 29th has a new etag, so the loader picks it up along with the 30th, and skips the unchanged 28th: skipped: 1.
Why not last-modified windows? Late files and clock gaps make them miss files or load them twice. A log is exact. Databricks Auto Loader keeps its list in a checkpoint, but skips a re-sent file unless you allow overwrites with allowOverwrites.
A rerun replaces and never appends: each source file overwrites its own output. Run one loader at a time, or write the log with an etag condition.
Scenario 7: Hundreds of partner files arrive every day and each must be validated and loaded within ten minutes. How do you build it?
Event driven, never a nightly batch. Partners upload into their own landing folders with a write-only SAS. Event Grid fires for each complete file, a queue buffers the events, and a small function validates and loads each file.
- Partners - write-only SAS
landing/- one folder each- Event Grid - complete files
- Queue - buffers events
- Validator - one file per event
- raw - good files
- quarantine - reason in metadata
The validator checks three things:
- Header -
order_id,partner,amount - Rows - at least one
- Amount - a number on every row
Good files go to the raw zone. Bad files go to quarantine with the reason in their metadata, and the partner gets a message. partners_drop.py has three partners drop four files, and globex uses semicolons. validate.py treats each file the way the function would per event. We run python partners_drop.py, then python validate.py.
from common import container
H = "order_id,partner,amount\n"
files = {"acme/0915.csv": H + "1,acme,20.5\n2,acme,7\n",
"acme/0930.csv": H + "3,acme,abc\n",
"globex/0920.csv": "id;amount\n7;12.5\n",
"initech/0925.csv": H + "9,initech,40\n"}
landing = container("landing")
for name, text in files.items():
landing.upload_blob(f"partners/{name}", text, overwrite=True)
print("partners dropped", len(files), "files into landing/partners/")from common import container
HEADER = "order_id,partner,amount"
landing = container("landing")
raw, quarantine = container("raw"), container("quarantine")
def problem(text):
head, *rows = text.strip().splitlines() or [""]
if head != HEADER:
return "bad header"
try:
[float(r.split(",")[2]) for r in rows]
except (IndexError, ValueError):
return "bad amount"
return None if rows else "no rows"
for b in landing.list_blobs(name_starts_with="partners/"):
src = landing.get_blob_client(b.name)
why = problem(src.download_blob(encoding="utf-8").readall())
dst = (quarantine if why else raw).get_blob_client(b.name)
dst.start_copy_from_url(src.url, metadata={"check": why or "ok"})
if dst.get_blob_properties().copy.status == "success":
src.delete_blob() # landing stays empty
print(f"{b.name:<26}", f"quarantine: {why}" if why else "raw")
print("left in landing:", len(list(
landing.list_blob_names(name_starts_with="partners/"))))$ python partners_drop.py
partners dropped 4 files into landing/partners/
$ python validate.py
partners/acme/0915.csv raw
partners/acme/0930.csv quarantine: bad amount
partners/globex/0920.csv quarantine: bad header
partners/initech/0925.csv raw
left in landing: 0acme/0915.csv and initech/0925.csv pass and land in raw. acme/0930.csv has a bad amount, and globex/0920.csv a bad header, so both go to quarantine. A file leaves landing only after its copy succeeds, and left in landing: 0.
To hold the ten minutes:
- Alert - a file waits in landing too long
- Dead-letter - events that can't be delivered
- Processed-files log - a retry never loads twice
Key points to remember
- Do the transfer math: AzCopy over the network, Data Factory for pipelines, Data Box for slow links
- Event Grid starts loads; filter data.api so only complete files trigger
- The change feed isn't supported with a hierarchical namespace: use events, inventory or a listing diff
- Copy between accounts server side with a source SAS or Entra ID; no object replication on HNS
- Track processed files by name and etag, and make every load idempotent
The full script
This is a Python script that writes every file and runs every command above in order against Azurite: run pip install azure-storage-blob azure-storage-file-datalake pyarrow pandas first.
Show the full script that writes the files and runs every step
# ========================================================================
# ADLS Gen2 Interview Q&A #7 - Loading and moving data
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on Azurite 3.37.0 (Microsoft's open-source Azure Storage emulator, Blob service)
# with azure-storage-blob 12.31.1, pyarrow and pandas; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
# 1. npm install -g azurite then azurite-blob --skipApiVersionCheck --location ./azurite-data
# (or the Docker image mcr.microsoft.com/azure-storage/azurite), and leave it running on port 10000.
# --skipApiVersionCheck lets a newer azure-storage-blob talk to it.
# This video also uses a second local account: start Azurite with the environment variable
# AZURITE_ACCOUNTS="devstoreaccount1:Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;partnerlake:cGFydG5lcmxha2Vsb2NhbGtleWZvcmF6dXJpdGVkZW1vMDEyMzQ1Njc4OTAxMjM0NTY3OA=="
# 2. pip install azure-storage-blob azure-storage-file-datalake pyarrow pandas requests
# 3. Set AZURE_STORAGE_CONNECTION_STRING to "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
# and PARTNER_STORAGE_CONNECTION_STRING to "DefaultEndpointsProtocol=http;AccountName=partnerlake;AccountKey=cGFydG5lcmxha2Vsb2NhbGtleWZvcmF6dXJpdGVkZW1vMDEyMzQ1Njc4OTAxMjM0NTY3OA==;BlobEndpoint=http://127.0.0.1:10000/partnerlake;"
# then run python ADLS_Interview_QA_07_Code.py.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any local data
# files) there, and a second run on old data gives different output.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('common.py', 'import os\nfrom azure.core.exceptions import ResourceExistsError\nfrom azure.storage.blob import BlobServiceClient\n\nsvc = BlobServiceClient.from_connection_string(\n os.environ["AZURE_STORAGE_CONNECTION_STRING"])\n\n\ndef container(name, client=svc):\n """Create the container if it is missing, return its client."""\n try:\n return client.create_container(name)\n except ResourceExistsError:\n return client.get_container_client(name)\n')
# Step 1: Do the transfer math: 10 TB over a 100 Mbps, 1 Gbps and 10 Gbps link
write('transfer_time.py', """\
# How long does 10 TB take over the network at full line rate?
SIZE_TB = 10
bits = SIZE_TB * 10**12 * 8 # 1 TB = 10**12 bytes, not TiB
for link, bps in [("100 Mbps", 100e6), ("1 Gbps", 1e9),
("10 Gbps", 10e9)]:
hours = bits / bps / 3600
print(f"{SIZE_TB} TB over {link:>8}: {hours:6.1f} hours"
f" = {hours / 24:4.1f} days")
""")
run('python transfer_time.py')
# Step 2: A file lands, then an event handler gets a CreateFile event and a FlushWithClose event
write('land.py', 'from common import container\n\ncsv = "order_id,amount\\n101,20.5\\n102,7.25\\n103,12\\n"\ncontainer("landing").upload_blob("acme/orders.csv", csv,\n overwrite=True)\nprint("partner upload: landing/acme/orders.csv,", len(csv), "bytes")\n')
write('event_create.json', """\
{
"eventType": "Microsoft.Storage.BlobCreated",
"subject":
"/blobServices/default/containers/landing/blobs/acme/orders.csv",
"data": {
"api": "CreateFile",
"contentLength": 0,
"blobType": "BlockBlob"
}
}
""")
write('event_flush.json', """\
{
"eventType": "Microsoft.Storage.BlobCreated",
"subject":
"/blobServices/default/containers/landing/blobs/acme/orders.csv",
"data": {
"api": "FlushWithClose",
"contentLength": 41,
"blobType": "BlockBlob"
}
}
""")
write('handler.py', """\
import json, sys
from common import svc
READY = {"PutBlob", "PutBlockList", "CopyBlob", "FlushWithClose"}
event = json.load(open(sys.argv[1]))
api = event["data"]["api"]
if api not in READY: # CreateFile: still empty
print(f"skip {api}: {event['data']['contentLength']} bytes")
sys.exit(0)
path = event["subject"].split("/containers/", 1)[1]
container, name = path.split("/blobs/", 1)
blob = svc.get_blob_client(container, name)
lines = blob.download_blob(encoding="utf-8").readall().splitlines()
print(f"load {container}/{name}: {len(lines) - 1} rows")
""")
run('python land.py')
run('python handler.py event_create.json')
run('python handler.py event_flush.json')
# Step 3: Which blobs changed since yesterday? A listing diff of name and etag
write('seed_raw.py', 'from common import container\n\nraw = container("raw")\nfor day in ["2026-10-26", "2026-10-27", "2026-10-28"]:\n raw.upload_blob(f"sales/{day}.csv", f"day,amount\\n{day},100\\n",\n overwrite=True)\nprint("raw/sales: 3 files")\n')
write('snapshot.py', """\
import json, sys
from common import container
raw = container("raw")
state = {b.name: b.etag # name -> etag
for b in raw.list_blobs(name_starts_with="sales/")}
json.dump(state, open(sys.argv[1], "w"))
print("snapshot", sys.argv[1] + ":", len(state), "blobs")
""")
write('todays_work.py', 'from common import container\n\nraw = container("raw")\nraw.upload_blob("sales/2026-10-28.csv",\n "day,amount\\n2026-10-28,120\\n", overwrite=True) # fix\nraw.upload_blob("sales/2026-10-29.csv", "day,amount\\n2026-10-29,90\\n")\nraw.delete_blob("sales/2026-10-26.csv") # cleaned up\nprint("updated 1, added 1, deleted 1")\n')
write('diff.py', """\
import json
old = json.load(open("yesterday.json"))
new = json.load(open("today.json"))
print("new :", sorted(new.keys() - old.keys()))
print("changed:", sorted(k for k in new.keys() & old.keys()
if new[k] != old[k])) # etag moved
print("deleted:", sorted(old.keys() - new.keys()))
""")
run('python seed_raw.py')
run('python snapshot.py yesterday.json')
run('python todays_work.py')
run('python snapshot.py today.json')
run('python diff.py')
# Step 4: Copy a blob from the partnerlake account into our lake, without and with a source SAS
write('copy_accounts.py', 'import os, time\nfrom datetime import datetime, timedelta, timezone\nfrom azure.core.exceptions import HttpResponseError\nfrom azure.storage.blob import (BlobServiceClient, BlobSasPermissions,\n generate_blob_sas)\nfrom common import container\n\npartner = BlobServiceClient.from_connection_string(\n os.environ["PARTNER_STORAGE_CONNECTION_STRING"])\nsrc = container("exports", partner).upload_blob(\n "orders/2026-10-29.csv", "order_id,amount\\n1,20.5\\n2,7.25\\n")\ndst = container("raw").get_blob_client("partner/2026-10-29.csv")\ntry:\n dst.start_copy_from_url(src.url) # no SAS\nexcept HttpResponseError as e:\n print("copy without SAS:", e.error_code.value)\nsas = generate_blob_sas(\n partner.account_name, "exports", src.blob_name,\n account_key=partner.credential.account_key,\n permission=BlobSasPermissions(read=True),\n expiry=datetime.now(timezone.utc) + timedelta(hours=1))\ndst.start_copy_from_url(f"{src.url}?{sas}") # server-side copy\nwhile dst.get_blob_properties().copy.status == "pending":\n time.sleep(5) # Azure: can be async\nprops = dst.get_blob_properties()\nprint("copy with SAS:", props.copy.status, props.size, "bytes")\n')
run('python copy_accounts.py')
# Step 5: A processed-files log: name and etag per file, rerun, then a re-sent file
write('drop_files.py', 'import sys\nfrom common import container\n\nlanding = container("landing")\nfor day in sys.argv[1:]: # the partner (re)sends\n landing.upload_blob(f"orders/{day}.csv",\n f"order_id,day\\n1,{day}\\n", overwrite=True)\nprint("partner sent:", ", ".join(sys.argv[1:]))\n')
write('load.py', """\
import json
from azure.core.exceptions import ResourceNotFoundError
from common import container
landing, control = container("landing"), container("control")
log = control.get_blob_client("orders_loaded.json")
try:
done = json.loads(log.download_blob().readall())
except ResourceNotFoundError:
done = {} # first run: empty log
loaded, skipped = [], 0
for b in landing.list_blobs(name_starts_with="orders/"):
if done.get(b.name) == b.etag: # same file, same version
skipped += 1
continue
# ... load b into curated, overwriting its own output ...
done[b.name] = b.etag
loaded.append(b.name)
log.upload_blob(json.dumps(done), overwrite=True)
print("loaded:", loaded, "| skipped:", skipped)
""")
run('python drop_files.py 2026-10-28 2026-10-29')
run('python load.py')
run('python load.py')
run('python drop_files.py 2026-10-29 2026-10-30')
run('python load.py')
# Step 6: Partner files: validate each one, good files to raw, bad files to quarantine
write('partners_drop.py', 'from common import container\n\nH = "order_id,partner,amount\\n"\nfiles = {"acme/0915.csv": H + "1,acme,20.5\\n2,acme,7\\n",\n "acme/0930.csv": H + "3,acme,abc\\n",\n "globex/0920.csv": "id;amount\\n7;12.5\\n",\n "initech/0925.csv": H + "9,initech,40\\n"}\nlanding = container("landing")\nfor name, text in files.items():\n landing.upload_blob(f"partners/{name}", text, overwrite=True)\nprint("partners dropped", len(files), "files into landing/partners/")\n')
write('validate.py', """\
from common import container
HEADER = "order_id,partner,amount"
landing = container("landing")
raw, quarantine = container("raw"), container("quarantine")
def problem(text):
head, *rows = text.strip().splitlines() or [""]
if head != HEADER:
return "bad header"
try:
[float(r.split(",")[2]) for r in rows]
except (IndexError, ValueError):
return "bad amount"
return None if rows else "no rows"
for b in landing.list_blobs(name_starts_with="partners/"):
src = landing.get_blob_client(b.name)
why = problem(src.download_blob(encoding="utf-8").readall())
dst = (quarantine if why else raw).get_blob_client(b.name)
dst.start_copy_from_url(src.url, metadata={"check": why or "ok"})
if dst.get_blob_properties().copy.status == "success":
src.delete_blob() # landing stays empty
print(f"{b.name:<26}", f"quarantine: {why}" if why else "raw")
print("left in landing:", len(list(
landing.list_blob_names(name_starts_with="partners/"))))
""")
run('python partners_drop.py')
run('python validate.py')
Part 8: Using the Lake From Analytics Engines
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on Azurite 3.37.0, Microsoft's open-source Azure Storage emulator, with azure-storage-blob 12.31.1, pyarrow and pandas. Features the emulator does not support are shown as code. Part of the Real Time Azure Data Lake Storage Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Azure Storage documentation (learn.microsoft.com/azure/storage) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.