Siemens Energy: event-driven, AI-aangedreven datapipeline op AWS

Klant: Siemens Energy · Data-afdelingUnknown NodeSector: Energie · ProductieUnknown NodeDiensten: Cloudarchitectuur, serverloze ontwikkeling, AI/ML-integratie, Infrastructure as Code (IaC)Unknown NodeAWS-services: S3, S3 Access Grants, Lambda, Step Functions, EventBridge, API Gateway, Bedrock Data Automation, Amplify, SQS, CloudFormation (CDK in Python)

De uitdaging

De data-afdeling van Siemens Energy beheert enorme hoeveelheden productmeetdata, zoals documenten, afbeeldingen, audio-opnames en videobestanden die gegenereerd zijn door wereldwijde productieactiviteiten, turbines en andere installaties en apparatuur.

Het team had een systeem nodig dat:

• Bestanden van willekeurige grootte kon verwerken, van kilobytes tot multi-gigabyte meetdatasets en beelden met hoge resolutie zonder payloadlimieten te bereiken of prestaties te verminderen.
• Automatisch rijke metadata kon extraheren uit elk geüpload bestand, ongeacht de modaliteit (tekst, afbeelding, audio, video), met behulp van AI. Wanneer bepaalde metadata vooraf bekend waren, moest het mogelijk zijn deze samen te voegen met de automatisch gegenereerde metadata.
• Toegangscontrole op partitieniveau kon afdwingen, zodat verschillende teams en gebruikers alleen data binnen hun aangewezen bereik konden zien en wijzigen, geïntegreerd met de bestaande Microsoft Entra ID (Azure AD) SSO-identiteitsprovider van Siemens Energy.
• Kon opschalen naar honderdduizenden bestanden zonder de beperkingen van het levenscyclusbeheer van tools zoals OneDrive of SharePoint (~300.000 bestandenplafond).
• Een webgebaseerde interface kon bieden voor niet-technische gebruikers om bestanden te uploaden, bladeren, downloaden en bekijken na authenticatie met bedrijfsbrede SSO.

Het systeem moest productierijp zijn, stresstests doorstaan en uitrolbaar zijn in meerdere omgevingen (sandbox, development, UAT, productie) met volledige Infrastructure as Code (IaC).

De oplossing

We hebben de Product Measurement Data Pipeline (PMDP) ontworpen en opgeleverd, een suite van onderling verbonden microservices en interfaces, volledig serverloos draaiend op AWS.

Architectuuroverzicht

Het platform bestaat uit:

1. File Manager API: een RESTful API-dienst voor CRUD-operaties op bestanden, multi-chunk uploads, generatie van vooraf ondertekende URL’s, metadata-operaties en partitiegebaseerde toegangscontrole.
2. Microservice voor metadataverrijking: een event-driven, multimodale AI-pipeline die automatisch gestructureerde metadata extraheert uit geüploade bestanden met Amazon Bedrock Data Automation.
3. Webapplicatie: een op AWS Amplify gehoste React-applicatie met een Storage Browser UI en Siemens Energy SSO-integratie.
4. Stresstestsuite: een CLI-middel voor het valideren van het systeem onder belasting met gelijktijdige uploads en downloads van bestanden tot meerdere gigabytes.

Alle infrastructuur is gedefinieerd in Python CDK, gevalideerd en automatisch uitgerold via CI/CD-pipelines (self-hosted GitLab).

Diepgaand: hoe het werkt

1. Hive-stijl objectsleutelpartitionering

Elk bestand dat via de API wordt geüpload, wordt opgeslagen in S3 met een Apache Hive-stijl partitioneringsschema. Dit is niet alleen organisatorisch. Het maakt hoogperformante queries mogelijk en is direct gekoppeld aan het toegangscontrolemodel.

# construct_object_key.py — Deterministische, bevraagbare S3-sleutelgeneratieUnknown NodeUnknown Nodeclass Capability(str, Enum):Unknown Node EDAA = "edaa"Unknown Node CUSTOMER_FACING = "customer-facing"Unknown Node MANUFACTURING = "manufacturing"Unknown Node QUALITY = "quality"Unknown Node # ...Unknown NodeUnknown Nodedef construct_object_key(Unknown Node capability: Capability | str | None = None,Unknown Node file_name: str = "",Unknown Node sub_capability: str | None = None,Unknown Node year: str | None = None,Unknown Node month: str | None = None,Unknown Node day: str | None = None,Unknown Node) -> str:Unknown Node now = datetime.now()Unknown Node generated_hash = hashlib.md5(Unknown Node f"{capability}{sub_capability}{file_name}".encode()Unknown Node ).hexdigest()[:2]Unknown NodeUnknown Node return (Unknown Node f"capability={capability}/"Unknown Node f"sub-capability={sub_capability or 'unknown'}/"Unknown Node f"year={year or now.year}/"Unknown Node f"month={month or f'{now.month:02d}'}/"Unknown Node f"day={day or f'{now.day:02d}'}/"Unknown Node f"hash={generated_hash}/"Unknown Node f"{file_name}"Unknown Node )

Een bestand geüpload als report.pdf onder de manufacturing-capability op 23 december 2025 wordt:

Deze structuur stelt AWS Athena, AWS Glue of elke Hive-compatibele tool in staat om het data lake efficiënt te bevragen op capability, datumbereik of elke combinatie van partitiesleutels. De tweetekenige hash voorkomt hotspots bij intensieve schrijfbewerkingen in S3.

2. S3 Access Grants met Entra ID-federatie

In plaats van IAM-beleid per gebruiker te beheren, hebben we S3 Access Grants geïmplementeerd — een relatief nieuwe AWS-functie die claims van identiteitsproviders rechtstreeks koppelt aan S3-prefix-niveau-machtigingen.

Elke Entra ID-gebruiker of -groep wordt gekoppeld aan een specifieke S3-partitie. Wanneer een gebruiker zich authenticeert, wordt de oid-claim uit het JWT-token gebruikt om de bijbehorende IAM-rol op te zoeken, die vervolgens wordt aangenomen via sts:AssumeRoleWithWebIdentity. De rol is beperkt tot de partitie van de gebruiker via een Access Grant.

# access_grants.py — Per-gebruiker partitie-isolatie via S3 Access GrantsUnknown NodeUnknown NodeCfnAccessGrant(Unknown Node self,Unknown Node f"AccessGrantUser{user_idx}",Unknown Node access_grants_location_id=user_location.ref,Unknown Node permission="READWRITE",Unknown Node grantee=CfnAccessGrant.GranteeProperty(Unknown Node grantee_identifier=user_role.role_arn,Unknown Node grantee_type="IAM",Unknown Node ),Unknown Node)

De Lambda-integratie wisselt vervolgens het Entra ID-token in voor beperkte AWS-referenties bij elk verzoek, met caching om overbodige STS-aanroepen te vermijden:

# get_cached_or_exchange_credentials.py — Tokenuitwisseling met cachingUnknown NodeUnknown Nodedef get_cached_or_exchange_credentials(id_token: str) -> CredentialsTypeDef:Unknown Node key = _cache_key_from_token(id_token)Unknown Node now = time.time()Unknown Node with _CACHE_LOCK:Unknown Node entry = _CACHE.get(key)Unknown Node if entry and entry["expires_at"] > now + _SKEW_SECONDS:Unknown Node return entry["credentials"]Unknown NodeUnknown Node new_credentials = _exchange_and_assume_with_expiry(id_token)Unknown NodeUnknown Node with _CACHE_LOCK:Unknown Node _CACHE[key] = {Unknown Node "credentials": new_credentials,Unknown Node "expires_at": new_credentials["Expiration"].timestamp(),Unknown Node }Unknown Node return new_credentialsUnknown Node

Dit betekent dat Gebruiker A in de manufacturing-partitie geen bestanden kan lezen of schrijven in de quality-partitie van Gebruiker B. Dit wordt afgedwongen op S3-niveau, niet alleen op applicatieniveau.

3. Multimodale AI-metadataverrijking

Wanneer een bestand in S3 terechtkomt, start automatisch een event-driven pipeline. Het systeem detecteert het bestandstype, genereert een op maat gemaakte Bedrock Data Automation-blueprint, voert de extractietaak uit en publiceert de gestructureerde resultaten terug naar EventBridge.

De volledige workflow wordt georkestreerd door Step Functions met JSONata-expressies:

# workflow.py — Step Functions-orkestratie met Bedrock Data AutomationUnknown NodeUnknown Nodedefinition_body = sfn.DefinitionBody.from_chainable(Unknown Node extract_event_dataUnknown Node .next(generate_blueprint) # Dynamisch blueprint op basis van metadatavereistenUnknown Node .next(start_data_automation_job) # Bedrock Data AutomationUnknown Node .next(wait_for_job_completion) # Asynchroon wachten met task tokenUnknown Node .next(normalize_event_data) # EventBridge Schema-gevalideerde outputUnknown Node .next(publish_output_event) # EventBridge voltooiingseventUnknown Node .next(sfn.Succeed(self, "FileMetadataEnrichmentCompletion"))Unknown Node)

De blueprintgenerator maakt dynamische extractieschema’s aan op basis van de bestandsmodaliteit en de door de gebruiker opgegeven metadatavereisten. Een afbeelding krijgt bounding box-detectie en categorisatie. Een document krijgt een samenvatting en extractie van kernpunten. Audio krijgt transcriptie en sprekeridentificatie:

# bedrock_data_automation_blueprint_generator.pyUnknown NodeUnknown Nodedef generate_blueprint_schema(enrichments, blueprint_type):Unknown Node properties = {}Unknown NodeUnknown Node # Basiseigenschappen — worden altijd geëxtraheerdUnknown Node properties["data_classification"] = {Unknown Node "type": "string",Unknown Node "instruction": "The data classification level (public, internal, confidential, restricted)",Unknown Node }Unknown Node properties["summary"] = {Unknown Node "type": "string",Unknown Node "instruction": "A brief summary of the file content",Unknown Node }Unknown Node properties["keywords"] = {Unknown Node "type": "array",Unknown Node "items": {"type": "string"},Unknown Node "instruction": "Key terms and keywords extracted from the document",Unknown Node }Unknown NodeUnknown Node # Modaliteitsspecifieke verrijkingenUnknown Node if enrichments.get("audio", {}).get("transcribe"):Unknown Node properties["transcript"] = {Unknown Node "type": "string",Unknown Node "instruction": "Full transcript of the audio content",Unknown Node }Unknown Node if enrichments.get("video", {}).get("scenes"):Unknown Node properties["scenes"] = {Unknown Node "type": "array",Unknown Node "items": {Unknown Node "type": "object",Unknown Node "properties": {Unknown Node "scene_number": {"type": "number"},Unknown Node "start_time": {"type": "number"},Unknown Node "end_time": {"type": "number"},Unknown Node "description": {"type": "string"},Unknown Node },Unknown Node },Unknown Node "instruction": "Scene changes detected in the video with timestamps",Unknown Node }Unknown NodeUnknown Node return {"class": f"{blueprint_type.capitalize()}Metadata", "properties": properties}

De verrijkte metadata wordt naast het originele bestand opgeslagen als een .metadata.json-sidecarbestand op dezelfde Hive-gepartitioneerde locatie, waardoor het direct bevraagbaar is.

4. Ondersteuning voor multi-gigabyte uploads

API Gateway heeft een payloadlimiet van 10 MB. Productmeetbestanden kunnen gigabytes groot zijn. We hebben dit opgelost met een op vooraf ondertekende URL’s gebaseerde multipart-uploadstroom die API Gateway volledig omzeilt voor het zware werk.

De API berekent optimale chunkgroottes, initieert een multipart-upload en retourneert vooraf ondertekende URL’s voor elke chunk. De client uploadt rechtstreeks naar S3:

# multi_chunk.py — Orkestratie van multipart-upload met vooraf ondertekende URL'sUnknown NodeUnknown Nodedef initiate_multi_chunk_upload_presigned(body):Unknown Node chunk_size, num_chunks = _calculate_chunk_size(request.fileSize)Unknown NodeUnknown Node response = s3.create_multipart_upload(Unknown Node Bucket=bucket, Key=key, ServerSideEncryption="aws:kms"Unknown Node )Unknown NodeUnknown Node presigned_urls = []Unknown Node for chunk_number in range(1, num_chunks + 1):Unknown Node presigned_url = s3.generate_presigned_url(Unknown Node "upload_part",Unknown Node Params={Unknown Node "Bucket": bucket, "Key": key,Unknown Node "UploadId": response["UploadId"],Unknown Node "PartNumber": chunk_number,Unknown Node },Unknown Node ExpiresIn=expires_in,Unknown Node )Unknown Node presigned_urls.append({Unknown Node "chunkNumber": chunk_number,Unknown Node "url": presigned_url,Unknown Node "startByte": (chunk_number - 1) * chunk_size,Unknown Node "endByte": min(chunk_number * chunk_size - 1, request.fileSize - 1),Unknown Node })Unknown NodeUnknown Node return {"uploadId": response["UploadId"], "chunks": presigned_urls}

Aan de frontendzijde handelt de webapplicatie dit transparant af — kleine bestanden gaan via de API, grote bestanden schakelen automatisch over naar multipart:

// api.service.ts — Automatische selectie van uploadstrategieUnknown NodeUnknown Nodeasync upload(file: File, request: InitiateUploadRequest, onProgress?) {Unknown Node if (file.size > FILE_SIZE_THRESHOLD) {Unknown Node return this.multiChunkUploadService.uploadFile(file, request, { onProgress });Unknown Node }Unknown NodeUnknown Node const base64Content = await readFileAsBase64(file);Unknown Node return this.httpService.request('/files', 'POST', {Unknown Node body: { content: base64Content, fileName: request.fileName },Unknown Node });Unknown Node}

5. De webapplicatie

De frontend is een React-applicatie gebouwd op de Storage Browser-component van AWS Amplify, aangepast met acties die via onze API worden gerouteerd in plaats van rechtstreeks naar S3. Dit geeft ons volledige controle over toegangscontrole, metadata-operaties en uploadstrategieën, terwijl het een gepolijste, vertrouwde bestandsbeheerervaring biedt.

// storage-browser.provider.tsx — Aangepaste Storage Browser met API-gestuurde actiesUnknown NodeUnknown Nodeconst { StorageBrowser } = createStorageBrowser({Unknown Node config: {Unknown Node registerAuthListener: async (onAuthStateChange) => {Unknown Node const authService = getAuthService();Unknown Node authService.registerAuthListener(onAuthStateChange);Unknown Node },Unknown Node listLocations: async ({ options }) => {Unknown Node return await apiService.getLocations({Unknown Node options: { pageSize: 30, nextToken: options?.nextToken },Unknown Node });Unknown Node },Unknown Node },Unknown Node actions: actionsBuilder.buildActions(),Unknown Node});

Gebruikers loggen in met hun Siemens Energy Entra ID-referenties en zien onmiddellijk alleen de partities waartoe ze toegang hebben. Ze kunnen bestanden van elke grootte uploaden, door de Hive-gepartitioneerde mappenstructuur bladeren, documenten en afbeeldingen inline bekijken, downloaden via vooraf ondertekende URL’s en metadata bewerken — allemaal zonder de browser te verlaten.

6. Cross-account event-driven architectuur

De File Manager en de microservice voor metadataverrijking draaien in afzonderlijke AWS-accounts. Wanneer een bestand wordt geüpload, triggert de S3-bucket van de File Manager een Lambda die een FileMetadataEnrichmentRequest-event publiceert naar een cross-account EventBridge-bus.

# file_metadata_enrichment_processor.py — Cross-account eventpublicatieUnknown NodeUnknown Nodedef put_event(bucket, key, size=None, etag=None, enrichments=None):Unknown Node detail = create_event_detail(bucket, key, size, etag, enrichments)Unknown NodeUnknown Node return events_client.put_events(Entries=[{Unknown Node "Source": "com.siemens-energy.pmdp.file-metadata-enrichment",Unknown Node "DetailType": "FileMetadataEnrichmentRequest",Unknown Node "Detail": json.dumps(detail),Unknown Node "EventBusName": EVENT_BUS_ARN, # Cross-account ARNUnknown Node }])

Aan de verrijkingszijde routeren EventBridge-regels events via SQS (met DLQ voor veerkracht) naar een EventBridge Pipe die het event valideert tegen een schemaregister voordat de Step Functions-workflow wordt aangeroepen. Voltooiings- en uitzonderingsevents worden teruggestuurd naar het oorspronkelijke account.

Deze ontkoppelde architectuur betekent dat de verrijkingsmicroservice hergebruikt kan worden door elk team bij Siemens Energy. Ze hoeven alleen events naar de bus te publiceren.

Stresstesten: bewijs op schaal

Het systeem moest enorme hoeveelheden data aankunnen, zowel inkomend als uitgaand. Sommige van hun bestandsrepositories konden dagen in beslag nemen om simpelweg te worden verwijderd. We hebben een speciale stresstest-CLI gebouwd die bestanden van configureerbare groottes genereert (van 100 KB tot 5 GB+), ze gelijktijdig uploadt, downloadt via vooraf ondertekende URL’s en de integriteit verifieert met MD5-checksums.

Testruns valideerden:

• Gelijktijdige uploads van 20+ bestanden tegelijk, inclusief multi-gigabyte payloads via multipart
• 100% slagingspercentage over honderden bestanden in een enkele testrun
• Downloadverificatie die byte-voor-byte integriteit bevestigt na een volledige rondgang door de hele pipeline
• Automatische opschoning van testartefacten uit zowel S3 als lokale opslag

Infrastructure as Code: alles in CDK

Het volledige platform, beide AWS-accounts, alle services, alle IAM-rollen, alle eventroutering zijn gedefinieerd in Python CDK. Omgevingsspecifieke configuratie wordt beheerd via Hydra/OmegaConf, waardoor het triviaal is om een nieuwe omgeving op te zetten of een nieuw team aan te sluiten.

# config.py — Typeveilige, omgevingsspecifieke configuratieUnknown NodeUnknown Nodeconfig_environment = make_config(Unknown Node env=zf(cdk.Environment),Unknown Node environment=zf(Literal["sandbox", "dev", "uat", "prd"]),Unknown Node s3explorer=zf(config_s3explorer),Unknown Node access_grants=zf(config_access_grants),Unknown Node project_name=zf(str, default="mfg-product-measurement-data-pipeline"),Unknown Node file_metadata_enrichment_event_bus_arn=zf(str),Unknown Node)

CI/CD voor de webapplicatie gebruikt GitLab OIDC-federatie. Geen langlevende referenties, geen secrets om te roteren. De CDK-stack provisioneert de OIDC-provider, de deploymentrol en de Amplify-bronbucket in één construct.

Resultaten

• Bestandsgrootte-ondersteuning: onbeperkt (getest tot 5 GB+)
• Upload-gelijktijdigheid: 20+ gelijktijdige uploads
• Metadata-extractie: automatisch voor documenten, afbeeldingen, audio en video
• Toegangscontrole: per-gebruiker en per-groep partitie-isolatie via S3 Access Grants
• Omgevingen: 4 (sandbox, development, UAT, productie) vanuit één CDK-codebase
• Identiteitsintegratie: Microsoft Entra ID SSO met OIDC-federatie
• Infrastructuur: 100% Infrastructure as Code (Python CDK)

Technologiestack

• Compute: AWS Lambda (Python 3.14, ARM64)
• Orkestratie: AWS Step Functions (JSONata)
• AI/ML: Amazon Bedrock Data Automation
• Opslag: Amazon S3 (Intelligent-Tiering, KMS-encryptie, Transfer Acceleration)
• API: Amazon API Gateway (REST) met Lambda Powertools + Swagger
• Events: Amazon EventBridge, EventBridge Pipes, SQS
• Identiteit: Microsoft Entra ID, OIDC-federatie, S3 Access Grants
• Frontend: React, Vite, AWS Amplify, Amplify UI Storage Browser
• IaC: AWS CDK (Python), Hydra/OmegaConf
• CI/CD: GitLab CI

Conclusie

Dit project vereiste diepgaande expertise over de gehele AWS-stack, van IAM-beleidsontwerp op laag niveau en cross-account eventroutering tot geavanceerde Bedrock Data Automation-integratie en aanpassing van de Amplify Storage Browser.

We hebben ons best gedaan om verder te gaan dan de oorspronkelijke specificaties en de nieuwste en beste AWS-cloudinnovaties aan te bevelen. Het resultaat was een systeem waarbij het uploaden van een bestand een AI-pipeline triggert die het verrijkt met gestructureerde metadata, het opslaat in een bevraagbare partitie en het direct beschikbaar maakt via een webinterface. Allemaal zonder dat de gebruiker iets anders hoeft te doen dan slepen en neerzetten.

Veel organisaties moeten robuuste, moderne cloudapplicaties bouwen op AWS; systemen die echte schaal aankunnen, integreren met enterprise-identiteitsproviders en AI inzetten waar het ertoe doet. Bent u er een van? Laten we praten.