Siemens Energy: подієво-орієнтований конвеєр даних на базі AI в AWS

Клієнт: Siemens Energy · Data Division
Галузь: Енергетика · Виробництво
Послуги: Хмарна архітектура, Serverless-розробка, інтеграція AI/ML, Infrastructure as Code (IaC)
Сервіси AWS: S3, S3 Access Grants, Lambda, Step Functions, EventBridge, API Gateway, Bedrock Data Automation, Amplify, SQS, CloudFormation (CDK на Python)

Виклик

Підрозділ даних Siemens Energy керує величезними обсягами вимірювальних даних продукції — документами, зображеннями, аудіозаписами та відеофайлами, що генеруються в рамках глобальних виробничих операцій, турбін та іншого обладнання.

Команді потрібна була система, яка могла б:

• Приймати файли довільного розміру — від кілобайтів до багатогігабайтних наборів вимірювальних даних та зображень високої роздільної здатності — без обмежень розміру або погіршення продуктивності.
• Автоматично витягувати розширені метадані з кожного завантаженого файлу, незалежно від модальності (текст, зображення, аудіо, відео), за допомогою AI. Водночас, коли певні метадані були відомі заздалегідь, потрібна була можливість об'єднати їх із згенерованими метаданими.
• Забезпечувати контроль доступу на рівні розділів, щоб різні команди та користувачі могли бачити й змінювати дані лише в межах свого визначеного обсягу, з інтеграцією з наявним провайдером ідентифікації Microsoft Entra ID (Azure AD) SSO від Siemens Energy.
• Масштабуватися до сотень тисяч файлів без обмежень управління життєвим циклом інструментів на кшталт OneDrive або SharePoint (ліміт ~300 000 файлів).
• Надавати вебінтерфейс для нетехнічних користувачів для завантаження, перегляду, скачування та попереднього перегляду файлів після автентифікації через корпоративний SSO.

Система мала бути промислового рівня, стрес-тестованою та готовою до розгортання в кількох середовищах (sandbox, development, UAT, production) із повною Infrastructure as Code (IaC).

Рішення

Ми спроєктували та реалізували Product Measurement Data Pipeline (PMDP) — набір взаємопов'язаних мікросервісів та інтерфейсів, що повністю працюють у serverless-режимі на AWS.

Огляд архітектури

Платформа складається з:

1. File Manager API: RESTful-сервіс, що обробляє CRUD-операції з файлами, багаточастинне завантаження, генерацію presigned URL, операції з метаданими та контроль доступу на рівні розділів.
2. File Metadata Enrichment Microservice: подієво-орієнтований мультимодальний AI-конвеєр, який автоматично витягує структуровані метадані із завантажених файлів за допомогою Amazon Bedrock Data Automation.
3. Вебзастосунок: React-застосунок, розміщений на AWS Amplify, що надає UI Storage Browser з інтеграцією SSO від Siemens Energy.
4. Набір стрес-тестів: CLI-інструмент для валідації системи під навантаженням з одночасним завантаженням та скачуванням файлів розміром до кількох гігабайтів.

Уся інфраструктура визначена в Python CDK, валідована та автоматично розгортається через CI/CD-конвеєри (self-hosted GitLab).

Детальний розбір: як це працює

1. Партиціонування ключів об'єктів у стилі Hive

Кожен файл, завантажений через API, зберігається в S3 за схемою партиціонування у стилі Apache Hive. Це не просто організаційне рішення — воно забезпечує високопродуктивні запити та безпосередньо відповідає моделі контролю доступу.

# construct_object_key.py — Детерміноване, придатне для запитів генерування ключів S3

   class Capability(str, Enum):
       EDAA = "edaa"
       CUSTOMER_FACING = "customer-facing"
       MANUFACTURING = "manufacturing"
       QUALITY = "quality"
       # ...

   def construct_object_key(
       capability: Capability | str | None = None,
       file_name: str = "",
       sub_capability: str | None = None,
       year: str | None = None,
       month: str | None = None,
       day: str | None = None,
   ) -> str:
       now = datetime.now()
       generated_hash = hashlib.md5(
           f"{capability}{sub_capability}{file_name}".encode()
       ).hexdigest()[:2]

       return (
           f"capability={capability}/"
           f"sub-capability={sub_capability or 'unknown'}/"
           f"year={year or now.year}/"
           f"month={month or f'{now.month:02d}'}/"
           f"day={day or f'{now.day:02d}'}/"
           f"hash={generated_hash}/"
           f"{file_name}"
       )

Файл, завантажений як report.pdf у розділі manufacturing 23 грудня 2025 року, перетворюється на:

capability=manufacturing/sub-capability=unknown/year=2025/month=12/day=23/hash=a3/report.pdf

Ця структура дозволяє Athena, Glue або будь-якому Hive-сумісному інструменту ефективно запитувати озеро даних за capability, діапазоном дат або будь-якою комбінацією ключів розділів. Двосимвольний хеш запобігає «гарячим точкам» при інтенсивному записі в S3.

2. S3 Access Grants з федерацією Entra ID

Замість управління IAM-політиками для кожного користувача ми впровадили S3 Access Grants — відносно нову функцію AWS, яка зіставляє claims провайдера ідентифікації безпосередньо з дозволами на рівні префіксів S3.

Кожен користувач або група Entra ID зіставляється з конкретним розділом S3. Під час автентифікації claim oid з JWT-токена використовується для пошуку IAM-ролі, яка потім отримується через sts:AssumeRoleWithWebIdentity. Роль обмежена їхнім розділом через Access Grant.

   # access_grants.py — Ізоляція розділів для кожного користувача через S3 Access Grants

   CfnAccessGrant(
       self,
       f"AccessGrantUser{user_idx}",
       access_grants_location_id=user_location.ref,
       permission="READWRITE",
       grantee=CfnAccessGrant.GranteeProperty(
           grantee_identifier=user_role.role_arn,
           grantee_type="IAM",
       ),
   )

Інтеграція з Lambda обмінює токен Entra ID на обмежені AWS-облікові дані при кожному запиті, з кешуванням для уникнення зайвих STS-викликів:

# get_cached_or_exchange_credentials.py — Обмін токенами з кешуванням
 
def get_cached_or_exchange_credentials(id_token: str) -> CredentialsTypeDef:
    key = _cache_key_from_token(id_token)
    now = time.time()
    with _CACHE_LOCK:
        entry = _CACHE.get(key)
        if entry and entry["expires_at"] > now + _SKEW_SECONDS:
            return entry["credentials"]
 
    new_credentials = _exchange_and_assume_with_expiry(id_token)
 
    with _CACHE_LOCK:
        _CACHE[key] = {
            "credentials": new_credentials,
            "expires_at": new_credentials["Expiration"].timestamp(),
        }
    return new_credentials

Це означає, що Користувач A у розділі manufacturing не може читати чи записувати файли в розділі Користувача B — quality. Це забезпечується на рівні S3, а не лише на рівні застосунку.

3. Мультимодальне збагачення метаданих за допомогою AI

Коли файл потрапляє в S3, автоматично запускається подієво-орієнтований конвеєр. Система визначає тип файлу, генерує відповідний blueprint Bedrock Data Automation, запускає завдання з вилучення даних та публікує структуровані результати назад у EventBridge.

Весь робочий процес оркеструється Step Functions із використанням виразів JSONata:

# workflow.py — Оркестрація Step Functions із Bedrock Data Automation
 
definition_body = sfn.DefinitionBody.from_chainable(
    extract_event_data
    .next(generate_blueprint)          # Динамічний blueprint на основі вимог до метаданих
    .next(start_data_automation_job)   # Bedrock Data Automation
    .next(wait_for_job_completion)     # Асинхронне очікування з токеном завдання
    .next(normalize_event_data)        # Вивід EventBridge, перевірений за схемою
    .next(publish_output_event)        # Подія завершення EventBridge
    .next(sfn.Succeed(self, "FileMetadataEnrichmentCompletion"))
)

Генератор blueprints динамічно створює схеми вилучення на основі модальності файлу та користувацьких вимог до метаданих. Для зображення виконується визначення обмежувальних рамок та категоризація. Для документа — резюмування та вилучення ключових тез. Для аудіо — транскрипція та ідентифікація спікерів:

# bedrock_data_automation_blueprint_generator.py
 
def generate_blueprint_schema(enrichments, blueprint_type):
    properties = {}
 
    # Базові властивості — завжди витягуються
    properties["data_classification"] = {
        "type": "string",
        "instruction": "The data classification level (public, internal, confidential, restricted)",
    }
    properties["summary"] = {
        "type": "string",
        "instruction": "A brief summary of the file content",
    }
    properties["keywords"] = {
        "type": "array",
        "items": {"type": "string"},
        "instruction": "Key terms and keywords extracted from the document",
    }
 
    # Modality-specific enrichments
    if enrichments.get("audio", {}).get("transcribe"):
        properties["transcript"] = {
            "type": "string",
            "instruction": "Full transcript of the audio content",
        }
    if enrichments.get("video", {}).get("scenes"):
        properties["scenes"] = {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "scene_number": {"type": "number"},
                    "start_time": {"type": "number"},
                    "end_time": {"type": "number"},
                    "description": {"type": "string"},
                },
            },
            "instruction": "Scene changes detected in the video with timestamps",
        }
 
    return {"class": f"{blueprint_type.capitalize()}Metadata", "properties": properties}

Збагачені метадані зберігаються поруч з оригінальним файлом як sidecar-файл .metadata.json у тій самій Hive-партиційованій локації, що робить їх одразу доступними для запитів.

4. Підтримка завантаження багатогігабайтних файлів

API Gateway має обмеження payload у 10 МБ. Файли вимірювальних даних продукції можуть важити гігабайти. Ми вирішили це за допомогою потоку багаточастинного завантаження на основі presigned URL, який повністю обходить API Gateway для важкої роботи.

API обчислює оптимальний розмір частин, ініціює багаточастинне завантаження та повертає presigned URL для кожної частини. Клієнт завантажує безпосередньо в S3:

# multi_chunk.py — Оркестрація попередньо підписаного багаточастинного завантаження
 
def initiate_multi_chunk_upload_presigned(body):
    chunk_size, num_chunks = _calculate_chunk_size(request.fileSize)
 
    response = s3.create_multipart_upload(
        Bucket=bucket, Key=key, ServerSideEncryption="aws:kms"
    )
 
    presigned_urls = []
    for chunk_number in range(1, num_chunks + 1):
        presigned_url = s3.generate_presigned_url(
            "upload_part",
            Params={
                "Bucket": bucket, "Key": key,
                "UploadId": response["UploadId"],
                "PartNumber": chunk_number,
            },
            ExpiresIn=expires_in,
        )
        presigned_urls.append({
            "chunkNumber": chunk_number,
            "url": presigned_url,
            "startByte": (chunk_number - 1) * chunk_size,
            "endByte": min(chunk_number * chunk_size - 1, request.fileSize - 1),
        })
 
    return {"uploadId": response["UploadId"], "chunks": presigned_urls}

На фронтенді вебзастосунок обробляє це прозоро: малі файли проходять через API, великі автоматично перемикаються на багаточастинне завантаження:

// api.service.ts — Автоматичний вибір стратегії завантаження
 
async upload(file: File, request: InitiateUploadRequest, onProgress?) {
  if (file.size > FILE_SIZE_THRESHOLD) {
    return this.multiChunkUploadService.uploadFile(file, request, { onProgress });
  }
 
  const base64Content = await readFileAsBase64(file);
  return this.httpService.request('/files', 'POST', {
    body: { content: base64Content, fileName: request.fileName },
  });
}

5. Вебзастосунок

Фронтенд — це React-застосунок, побудований на компоненті Storage Browser від AWS Amplify, налаштований з діями, які маршрутизуються через наш API замість прямого звернення до S3. Це дає нам повний контроль над контролем доступу, операціями з метаданими та стратегіями завантаження, забезпечуючи при цьому відшліфований і звичний досвід управління файлами.

// storage-browser.provider.tsx — Кастомний Storage Browser з діями через API
 
const { StorageBrowser } = createStorageBrowser({
  config: {
    registerAuthListener: async (onAuthStateChange) => {
      const authService = getAuthService();
      authService.registerAuthListener(onAuthStateChange);
    },
    listLocations: async ({ options }) => {
      return await apiService.getLocations({
        options: { pageSize: 30, nextToken: options?.nextToken },
      });
    },
  },
  actions: actionsBuilder.buildActions(),
});

Користувачі входять за допомогою своїх облікових даних Siemens Energy Entra ID і одразу бачать лише ті розділи, до яких мають доступ. Вони можуть завантажувати файли будь-якого розміру, переглядати Hive-партиційовану структуру папок, попередньо переглядати документи та зображення, скачувати через presigned URL та редагувати метадані — і все це без виходу з браузера.

6. Кросакаунтна подієво-орієнтована архітектура

File Manager та мікросервіс Metadata Enrichment працюють у різних AWS-акаунтах. Коли файл завантажується, S3-бакет File Manager'а запускає Lambda, яка публікує подію FileMetadataEnrichmentRequest у кросакаунтну шину EventBridge.

# file_metadata_enrichment_processor.py — Публікація подій між акаунтами
 
def put_event(bucket, key, size=None, etag=None, enrichments=None):
    detail = create_event_detail(bucket, key, size, etag, enrichments)
 
    return events_client.put_events(Entries=[{
        "Source": "com.siemens-energy.pmdp.file-metadata-enrichment",
        "DetailType": "FileMetadataEnrichmentRequest",
        "Detail": json.dumps(detail),
        "EventBusName": EVENT_BUS_ARN,  # ARN між акаунтами
    }])

На стороні збагачення правила EventBridge маршрутизують події через SQS (з DLQ для стійкості) у EventBridge Pipe, який валідує подію за реєстром схем перед запуском робочого процесу Step Functions. Події завершення та виняткові ситуації пересилаються назад до вихідного акаунту.

Ця розв'язана архітектура означає, що мікросервіс збагачення може бути повторно використаний будь-якою командою в Siemens Energy. Їм просто потрібно публікувати події в шину.

Стрес-тестування: підтвердження масштабованості

Система мала обробляти величезні обсяги даних як на вхід, так і на вихід. Деякі їхні файлові репозиторії могли потребувати днів лише для видалення. Ми створили спеціалізований CLI для стрес-тестування, який генерує файли настроюваного розміру (від 100 КБ до 5 ГБ+), завантажує їх одночасно, скачує через presigned URL та перевіряє їхню цілісність за допомогою контрольних сум MD5.

Тестові запуски підтвердили:

• Одночасне завантаження 20+ файлів одночасно, включаючи багатогігабайтні payload через multipart
• 100% успішність серед сотень файлів за один тестовий запуск
• Верифікація скачування, що підтверджує побайтову цілісність після проходження через весь конвеєр
• Автоматичне очищення тестових артефактів як з S3, так і з локального сховища

Infrastructure as Code: все в CDK

Уся платформа — обидва AWS-акаунти, усі сервіси, усі IAM-ролі та вся маршрутизація подій — визначена в Python CDK. Конфігурація для конкретних середовищ управляється через Hydra/OmegaConf, що робить розгортання нового середовища або онбординг нової команди тривіальним.

# config.py — Типобезпечна конфігурація, специфічна для середовища
 
config_environment = make_config(
    env=zf(cdk.Environment),
    environment=zf(Literal["sandbox", "dev", "uat", "prd"]),
    s3explorer=zf(config_s3explorer),
    access_grants=zf(config_access_grants),
    project_name=zf(str, default="mfg-product-measurement-data-pipeline"),
    file_metadata_enrichment_event_bus_arn=zf(str),
)

CI/CD для вебзастосунку використовує GitLab OIDC-федерацію, без довготривалих облікових даних чи секретів, які потрібно ротувати. CDK-стек налаштовує OIDC-провайдер, роль для розгортання та source-бакет Amplify в єдиному конструкті.

Результати

• Підтримка розміру файлів: необмежений (протестовано до 5 ГБ+)
• Конкурентність завантаження: 20+ одночасних завантажень
• Вилучення метаданих: автоматичне для документів, зображень, аудіо та відео
• Контроль доступу: ізоляція розділів на рівні користувача та групи через S3 Access Grants
• Середовища: 4 (sandbox, development, UAT, production) з єдиної кодової бази CDK
• Інтеграція ідентифікації: Microsoft Entra ID SSO з OIDC-федерацією
• Інфраструктура: 100% Infrastructure as Code (Python CDK)

Технологічний стек

• Обчислення: AWS Lambda (Python 3.14, ARM64)
• Оркестрація: AWS Step Functions (JSONata)
• AI/ML: Amazon Bedrock Data Automation
• Сховище: Amazon S3 (Intelligent-Tiering, KMS-шифрування, Transfer Acceleration)
• API: Amazon API Gateway (REST) з Lambda Powertools + Swagger
• Події: Amazon EventBridge, EventBridge Pipes, SQS
• Ідентифікація: Microsoft Entra ID, OIDC-федерація, S3 Access Grants
• Фронтенд: React, Vite, AWS Amplify, Amplify UI Storage Browser
• IaC: AWS CDK (Python), Hydra/OmegaConf
• CI/CD: GitLab CI

Висновок

Цей проєкт вимагав глибокої експертизи в екосистемі AWS — від низькорівневого проєктування IAM-політик та кросакаунтної маршрутизації подій до інтеграції з передовою Bedrock Data Automation та кастомізації Amplify Storage Browser.

Ми зробили все можливе, щоб вийти за межі початкових специфікацій і рекомендувати найновіші та найкращі хмарні інновації AWS. Результатом стала система, в якій завантаження файлу запускає AI-конвеєр, що збагачує його структурованими метаданими, зберігає в розділі з можливістю запитів і робить одразу доступним для перегляду через вебінтерфейс — і все це без жодних дій користувача, окрім перетягування файлу.

Багатьом організаціям потрібно створювати надійні, сучасні хмарні застосунки на AWS — системи, які працюють у реальному масштабі, інтегруються з корпоративними провайдерами ідентифікації та використовують AI там, де це має значення. Ви одна з них? Давайте поговоримо.

Малік АлімухамедовІнженер, технооптиміст, підприємець, письменник, музикант, інвестор і мінімаліст. Я прагну автоматизувати якомога більше в своєму житті. Письмо допомагає мені кристалізувати думки та керувати ними. Технології — це мій хліб насущний, а також моя пристрасть. Ви можете сміливо звертатися будь-коли.