SFTP flat-file batch → event-driven ingestion

# Event-driven SFTP file ingestion — Lambda/Cloud Function pattern
# Triggered on S3 PUT (files transferred from SFTP) or equivalent event grid

def handle_file_event(event, ctx):
    bucket, key = event['bucket'], event['key']

    # 1. Read & validate file structure
    raw = s3.get_object(Bucket=bucket, Key=key)['Body'].read()
    records = parse_csv(raw)
    if not validate_header(records[0'], REQUIRED_COLS):
        raise ValueError(f"Invalid schema in {key}")

    # 2. Dedupe by content hash (stored in DynamoDB)
    content_hash = hashlib.sha256(raw).hexdigest()
    if dynamodb.get_item(Key={'hash': content_hash}).get('Item'):
        logger.info(f"Skipping duplicate {key}")
        return {"status": "duplicate_skipped"}

    # 3. Validate & publish each record as event
    for i, row in enumerate(records[1:], start=2):
        if is_valid(row):
            sns.publish(
                TopicArn=STREAM_TOPIC,
                Message=json.dumps({"source_file": key,
                                    "line": i,
                                    "data": row,
                                    "ts": datetime.utcnow().isoformat()})
            )

    # 4. Mark hash processed
    dynamodb.put_item(Item={'hash': content_hash, 'key': key, 'ts': datetime.utcnow().isoformat()})
    return {"status": "processed", "records": len(records)-1}

How we'd take this from discovery to a production-safe cutover — the phases, the canonical mapping, and the edge cases that bite.

Replace the fragile nightly SFTP flat-file batch with an event-driven pipeline that triggers on file arrival, validates against a defined schema, deduplicates using a state store, and publishes structured events downstream.

Implementation Phases

1. Audit existing batch jobs. Inventory all files landing on the SFTP server (/inbound/datafeed, /inbound/transactions, etc.). Document record counts, file naming conventions, and the downstream SQL inserts or API calls each job performs. This becomes your regression baseline.
2. Deploy an object store as the ingestion buffer. Configure an S3 bucket (sftp-land-zone) with a transfer prefix (/inbound/). Mirror SFTP writes using rsync in the interim, or wire the SFTP server to push directly to S3 if supported. Files now have a durable, versioned artifact with a checksum attribute.
3. Wire the event trigger. Attach an S3 event notification on sftp-land-zone to an EventBridge rule (sftp-file-arrival) that filters on *.csv and *.dat suffixes. The rule invokes a Lambda function (datafeed-file-trigger) or publishes to an SQS queue (datafeed-ingestion-queue) for durability.
4. Build the validation layer. In the triggered function, read the file header and enforce the schema contract. Validate field count, data types, and date formats against a schema registry (e.g., an OpenAPI spec or JSON Schema stored in /schemas/datafeed/v2.json). Reject non-conforming files to sftp-rejected/ and increment a validation_errors CloudWatch metric.
5. Implement deduplication. Before processing, compute the file checksum (SHA256) and query a DynamoDB table (file_processing_state) keyed on checksum. If the record exists with status = COMPLETED, skip processing and log a DUPLICATE_FILE event. Insert the checksum with status = IN_PROGRESS at transaction start; update to COMPLETED or FAILED on exit.
6. Transform and publish. Parse each record into a canonical JSON event with fields eventId, sourceFile, receivedAt, recordCount, and a payload array. Publish to an EventBridge custom bus (datafeed-events) or SNS topic (datafeed-ingested) for downstream consumers. Write a processing audit record to processing_audit_log.
7. Instrument observability. Emit CloudWatch metrics (file_received_count, processing_duration_seconds, record_published_count) and stream logs to a centralized aggregator. Configure alarms on dlq-message-count exceeding zero and on p95 processing latency above 60 seconds.
8. Staged cutover. Run the event-driven pipeline in shadow mode for two weeks, comparing output counts and sums against the legacy batch. Correct discrepancies. Once reconciled, switch downstream consumers to the new event bus, decommission the old SFTP-to-SQL jobs, and archive the legacy logs.

Canonical Field Mapping

Edge Cases

• Partial file arrival. A file appears but is still being written; S3 event fires prematurely. Guard by checking object size stability across two consecutive HEAD requests 5 seconds apart before processing.
• Duplicate file delivery. The upstream system retransmits a file with identical content. The dedup table catches this, but only if the checksum is computed before transformation to avoid false negatives from header timestamp differences.
• Schema drift. The sender introduces a new column without notice. The validation layer must reject and quarantine the file rather than silently dropping or misrouting records.
• Idempotency gaps. A function invocation fails after writing events but before updating the dedup table to COMPLETED. On retry, the in-flight record triggers a ConditionalCheckFailedException; handle it by re-querying status and resuming from the last committed offset.
• Poison messages. A malformed file causes repeated Lambda invocations and lands in the DLQ. Set a maximum receive count of 3 on the SQS queue and route to sftp-dead-letter/ with an alert.
• Clock skew on EFFECTIVE_DATE. Records with future-dated EFFECTIVE_DATE are valid upstream data but may cause downstream reconciliation failures. Tag these events with temporalAnomaly: true for manual review.

Cutover

Cutover must be fully reversible. Keep the legacy SFTP batch job scheduled and disabled (not deleted) for 30 days post-migration. During the transition window, run the event-driven pipeline in parallel: the batch job continues writing to the production database while the new pipeline writes to a shadow schema (staging_datafeed). Use a reconciliation query that compares record counts and checksummed sums grouped by sourceFile and effectiveDate between the two schemas. Any mismatch halts the cutover and pages the on-call engineer. Once the delta is zero for 14 consecutive days, promote the shadow schema, enable the DLQ alerts as production-grade, and archive the legacy job with a 90-day retention tag. If a rollback is required, re-enable the disabled cron schedule and repoint consumers to the original tables; the dedup table ensures no double-counting during the reversal window.