Siemens Energy : conduit de données IA piloté par les événements sur AWS

Client : Siemens Energy · Équipe de donnéesUnknown NodeSecteur : Énergie · Industrie manufacturièreUnknown NodeServices : Architecture cloud, développement serverless, intégration IA/ML, infrastructure sous forme de code (Infrastructure as Code, IaC)Unknown NodeServices AWS : S3, S3 Access Grants, Lambda, Step Functions, EventBridge, API Gateway, Bedrock Data Automation, Amplify, SQS, CloudFormation (CDK en Python)

Le défi

L'équipe de données de Siemens Energy gère d’énormes volumes de données de mesure de produit, telles que des documents, des images, des enregistrements audio et des fichiers vidéo, générés par les opérations de fabrication mondiales, ainsi que par les turbines et autres installations et équipements.

L’équipe avait besoin d’un système capable de :

• Ingérer des fichiers de taille arbitraire, des jeux de données de mesure de plusieurs gigaoctets jusqu’aux images en haute résolution. Le tout sans atteindre les limites de charge utile ni compromettre les performances.
• Extraire automatiquement des métadonnées riches à partir de chaque fichier téléchargé, quelle que soit la modalité (texte, image, audio ou vidéo), grâce à l’IA. Cependant, lorsque certaines métadonnées étaient connues à l’avance, il fallait pouvoir les fusionner avec celles générées automatiquement.
• Appliquer un contrôle d’accès au niveau des partitions afin que les différentes équipes et les utilisateurs puissent voir ou modifier uniquement les données dans leur périmètre désigné, intégré au fournisseur d’identité SSO Microsoft Entra ID (Azure AD) existant de Siemens Energy.
• Évoluer jusqu’à des centaines de milliers de fichiers sans les limitations de gestion du cycle de vie des outils comme OneDrive ou SharePoint (plafonnés à environ 300 000 fichiers).
• Fournir une interface web aux utilisateurs non techniques afin de leur permettre de télécharger, de parcourir et de prévisualiser des fichiers après authentification via le SSO de l’entreprise.

Le système devait être de qualité de production, testé en charge et déployable dans plusieurs environnements (sandbox, développement, UAT, production) avec une infrastructure sous forme de code (Infrastructure as Code, IaC) complète.

La Solution

Nous avons conçu et livré la Product Measurement Data Pipeline (PMDP), une suite de microservices et d’interfaces interconnectés, fonctionnant entièrement en mode serverless sur AWS.

Vue d’ensemble de l’architecture

La plateforme se compose de :

1. File Manager API : une API REST gérant les opérations CRUD sur les fichiers, les téléchargements en multi-morceau, la génération d’URL présignées, les opérations sur les métadonnées et le contrôle d’accès par partition.
2. Microservice d’enrichissement des métadonnées : un conduit de données multimodal, piloté par des événements, qui extrait automatiquement des métadonnées structurées à partir de fichiers téléchargés via Amazon Bedrock Data Automation.
3. Application web : une application React hébergée sur AWS Amplify, fournissant une interface Storage Browser avec l’intégration SSO de Siemens Energy.
4. Suite de tests de charge : un outil en ligne de commande (CLI) pour valider le système sous charge, avec des téléchargements et des téléversements simultanés de fichiers pouvant atteindre plusieurs gigaoctets.

Toute l’infrastructure est définie en Python CDK, validée et déployée automatiquement via des pipelines CI/CD (GitLab auto-hébergé).

En profondeur : comment ça fonctionne

1. Partitionnement des clés d’objets à la Hive

Chaque fichier téléchargé via l’API est stocké dans S3 selon un schéma de partitionnement conforme à celui d'Apache Hive. Ce n’est pas qu’organisationnel; cela permet des requêtes à haute performance et correspond directement au modèle de contrôle d’accès.

# construct_object_key.py — Génération déterministe de clés S3 interrogeablesUnknown 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 )

Un fichier téléchargé sous le nom report.pdf avec la capacité manufacturing le 23 décembre 2025 devient :

Cette structure permet à AWS Athena, à AWS Glue ou à tout outil compatible avec Hive d’interroger efficacement le lac de données par capacité, par plage de dates ou par toute combinaison de clés de partition. Le hash de deux caractères évite les points chauds (hotspots) lors d’écritures intensives sur S3.

2. S3 Access Grants avec fédération Entra ID

Plutôt que de gérer des politiques IAM par utilisateur, nous avons implémenté les S3 Access Grants. C'est une fonctionnalité AWS relativement récente qui associe directement les revendications du fournisseur d’identité aux autorisations au niveau des préfixes S3.

Chaque utilisateur ou groupe Entra ID est associé à une partition S3 spécifique. Lorsqu’un utilisateur s’authentifie, la revendication oid de son jeton JWT est utilisée pour rechercher son rôle IAM, qui est ensuite assumé via sts:AssumeRoleWithWebIdentity. Le rôle est limité à la partition de l’utilisateur via un Access Grant.

# access_grants.py — Isolation des partitions par utilisateur 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)

L’intégration Lambda échange ensuite le jeton Entra ID contre des identifiants AWS à portée limitée à chaque requête, avec mise en cache pour éviter les appels STS redondants :

# get_cached_or_exchange_credentials.py — Échange de jetons avec mise en cacheUnknown 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_credentials

Cela signifie que l’utilisateur A dans la partition manufacturing ne peut pas lire ou écrire des fichiers dans la partition quality de l’utilisateur B. Ceci est appliqué au niveau de S3, pas seulement au niveau applicatif.

3. Enrichissement multimodal des métadonnées par IA

Lorsqu’un fichier arrive dans S3, un conduit de données piloté par événements se déclenche automatiquement. Le système détecte le type de fichier, génère un blueprint Bedrock Data Automation adapté, exécute la tâche d’extraction et publie les résultats structurés vers EventBridge.

L’ensemble du workflow est orchestré par Step Functions à l’aide d’expressions JSONata :

# workflow.py — Orchestration Step Functions avec Bedrock Data AutomationUnknown NodeUnknown Nodedefinition_body = sfn.DefinitionBody.from_chainable(Unknown Node extract_event_dataUnknown Node .next(generate_blueprint) # Blueprint dynamique basé sur les exigences de métadonnéesUnknown Node .next(start_data_automation_job) # Bedrock Data AutomationUnknown Node .next(wait_for_job_completion) # Attente asynchrone avec task tokenUnknown Node .next(normalize_event_data) # Sortie validée par le schéma EventBridgeUnknown Node .next(publish_output_event) # Événement de complétion EventBridgeUnknown Node .next(sfn.Succeed(self, "FileMetadataEnrichmentCompletion"))Unknown Node)

Le générateur de blueprints génère dynamiquement des schémas d’extraction basés sur le type de fichier et sur les exigences de métadonnées spécifiées par l’utilisateur. Une image subit la détection de boîtes englobantes et la catégorisation. Un document permet d’obtenir la synthèse et l’extraction des points clés. L’audio obtient la transcription et l’identification des locuteurs :

# bedrock_data_automation_blueprint_generator.pyUnknown NodeUnknown Nodedef generate_blueprint_schema(enrichments, blueprint_type):Unknown Node properties = {}Unknown NodeUnknown Node # Propriétés de base — toujours extraitesUnknown 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 # Enrichissements spécifiques à la modalitéUnknown 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}

Les métadonnées enrichies sont stockées à côté du fichier original sous forme de fichier compagnon .metadata.json au même emplacement partitionné Hive, les rendant immédiatement interrogeables.

4. Prise en charge des téléchargements multi-gigaoctets

API Gateway a une limite de charge utile de 10 Mo. Les fichiers de mesure de produit peuvent peser des gigaoctets. Nous avons résolu ce problème avec un flux de téléchargement multipart basé sur des URL présignées, qui contourne entièrement l’API Gateway pour les opérations lourdes.

L’API calcule les tailles optimales des morceaux, initie un téléchargement en plusieurs parties et renvoie des URL présignées pour chaque morceau. Le client télécharge directement vers S3 :

# multi_chunk.py — Orchestration du téléchargement multipart avec URL présignéesUnknown 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}

Côté frontend, l’application web gère cela de manière transparente; les petits fichiers passent par l’API, les gros fichiers basculent automatiquement vers le multipart :

// api.service.ts — Sélection automatique de la stratégie de téléchargementUnknown 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. L’Application Web

Le frontend est une application React construite sur le composant Storage Browser d’AWS Amplify, personnalisée avec des actions qui passent par notre API plutôt que d’aller directement vers S3. Cela nous donne un contrôle total sur le contrôle d’accès, les opérations de métadonnées et les stratégies de téléchargement tout en offrant une expérience de gestion de fichiers soignée et familière.

// storage-browser.provider.tsx — Storage Browser personnalisé avec actions pilotées par l'APIUnknown 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});

Les utilisateurs se connectent avec leurs identifiants Entra ID Siemens Energy et voient immédiatement uniquement les partitions auxquelles ils ont accès. Ils peuvent télécharger des fichiers de toute taille, parcourir la structure de dossiers partitionnée conformément au standard Hive, prévisualiser des documents et des images en ligne, télécharger via des URL présignées et modifier les métadonnées, le tout sans quitter le navigateur.

6. Architecture événementielle inter-comptes

Le File Manager et le microservice d’enrichissement des métadonnées fonctionnent dans des comptes AWS distincts. Lorsqu’un fichier est téléchargé, le bucket S3 du File Manager déclenche une Lambda qui publie un événement FileMetadataEnrichmentRequest vers un bus EventBridge inter-comptes.

# file_metadata_enrichment_processor.py — Publication d'événements inter-comptesUnknown 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, # ARN inter-comptesUnknown Node }])

Côté enrichissement, les règles EventBridge acheminent les événements via SQS (avec DLQ pour la résilience) vers un EventBridge Pipe qui valide l’événement contre un registre de schémas avant d’invoquer le workflow Step Functions. Les événements de complétion et d’exception sont renvoyés vers le compte d’origine.

Cette architecture découplée permet au microservice d’enrichissement d’être réutilisé par n’importe quelle équipe de Siemens Energy. Il leur suffit de publier des événements sur le bus événementiel dédié.

Tests de charge : la preuve à l’échelle

Le système devait gérer d’énormes quantités de données, tant en entrée qu’en sortie. Certains de leurs dépôts de fichiers pouvaient prendre des jours rien que pour être supprimés. Nous avons construit un CLI de test de charge dédié qui génère des fichiers de tailles configurables (de 100 Ko à 5 Go+), les télécharge simultanément, les télécharge via des URL présignées et vérifie l’intégrité avec des sommes de contrôle MD5.

Les exécutions de tests ont validé :

• Téléchargements simultanés de 20+ fichiers en même temps, y compris des charges multi-gigaoctets via multipart
• Taux de réussite de 100 % sur des centaines de fichiers en une seule exécution de test
• Vérification des téléchargements confirmant l’intégrité octet par octet après un aller-retour complet à travers l’ensemble du pipeline
• Nettoyage automatique des artefacts de test depuis S3 et le stockage local

Infrastructure sous forme de code : tout en CDK

L’ensemble de la plateforme, les deux comptes AWS, tous les services, tous les rôles IAM et tout le routage des événements ont été définis dans le CDK en Python. La configuration spécifique à l’environnement est gérée via Hydra/OmegaConf, ce qui rend trivial le déploiement d’un nouvel environnement ou l’intégration d’une nouvelle équipe.

# config.py — Configuration typée et spécifique à l'environnementUnknown 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)

Le CI/CD de l’application web utilise la fédération OIDC de GitLab. Pas d’identifiants à longue durée de vie, pas de secrets à faire tourner. La stack CDK provisionne le fournisseur OIDC, le rôle de déploiement et le bucket source Amplify en un seul construct.

Résultats

• Taille de fichier supportée: illimitée (testée jusqu’à 5 Go+)
• Concurrence de téléchargement: 20+ téléchargements simultanés
• Extraction de métadonnées: automatique pour documents, images, audio et vidéo
• Contrôle d’accès: isolation des partitions par utilisateur et par groupe via S3 Access Grants
• Environnements: 4 (sandbox, développement, UAT, production) à partir d’une seule base de code CDK
• Intégration d’identité: Microsoft Entra ID SSO avec fédération OIDC
• Infrastructure: 100 % Infrastructure as Code (Python CDK)

Stack technologique

• Calcul : AWS Lambda (Python 3.14, ARM64)
• Orchestration : AWS Step Functions (JSONata)
• IA/ML : Amazon Bedrock Data Automation
• Stockage : Amazon S3 (Intelligent-Tiering, chiffrement KMS, Transfer Acceleration)
• API : Amazon API Gateway (REST) avec Lambda Powertools + Swagger
• Événements : Amazon EventBridge, EventBridge Pipes, SQS
• Identité : Microsoft Entra ID, fédération OIDC, S3 Access Grants
• Frontend : React, Vite, AWS Amplify, Amplify UI Storage Browser
• IaC : AWS CDK (Python), Hydra/OmegaConf
• CI/CD : GitLab CI

Conclusion

Ce projet a nécessité une expertise approfondie sur l’ensemble du stack AWS — de la conception de politiques IAM de bas niveau et du routage d’événements inter-comptes à l’intégration de pointe de Bedrock Data Automation et à la personnalisation de l’Amplify Storage Browser.

Nous avons fait de notre mieux pour aller au-delà des spécifications initiales et recommander les innovations cloud d’AWS les plus récentes et les plus performantes. Le résultat est un système où le téléchargement d’un fichier déclenche un pipeline d’IA qui l’enrichit de métadonnées structurées, le stocke dans une partition interrogeable et le rend immédiatement consultable via une interface web, le tout sans que l’utilisateur n’ait à faire autre chose que de glisser-déposer.

De nombreuses organisations ont besoin de construire des applications cloud robustes et modernes sur AWS ; des systèmes capables de gérer à grande échelle, d’intégrer les fournisseurs d’identité d’entreprise et d’exploiter l’IA là où elle est pertinente. Êtes-vous l’une d’entre elles ? Parlons-en.