**Approach (brief)** Use an AWS Lambda (Node.js) triggered by streaming source (Kinesis/Firehose). Process records in batches, stream-parse JSON, redact configured PII fields (deeply), buffer sanitized records and write compressed objects to S3 in parallel with controlled concurrency. Use retries, DLQ, and memory-aware chunking.**Pseudocode implementation**javascript
// Handler for Kinesis event
const AWS = require('aws-sdk');
const s3 = new AWS.S3();
const zlib = require('zlib');
const PII_FIELDS = ['email','ssn','credit_card'];
exports.handler = async (event) => {
const sanitized = [];
for (const rec of event.Records) {
try {
const json = JSON.parse(Buffer.from(rec.kinesis.data, 'base64').toString());
sanitized.push(redact(json));
} catch (e) {
console.error('parse error', e);
// Optionally send to DLQ
}
}
if (sanitized.length === 0) return;
const payload = Buffer.from(JSON.stringify(sanitized));
const gz = zlib.gzipSync(payload);
await s3.putObject({
Bucket: process.env.OUT_BUCKET,
Key: `logs/${Date.now()}.json.gz`,
Body: gz,
ContentEncoding: 'gzip',
ContentType: 'application/json'
}).promise();
};
function redact(obj){
if (obj === null || typeof obj !== 'object') return obj;
if (Array.isArray(obj)) return obj.map(redact);
const out = {};
for (const k of Object.keys(obj)){
if (PII_FIELDS.includes(k.toLowerCase())){
out[k] = '[REDACTED]';
} else {
out[k] = redact(obj[k]);
}
}
return out;
}
