AWS Study Group

Tích hợp metadata tùy chỉnh với Amazon S3 Metadata

bởi Bimal Gajjar và Carl Summers vào ngày 31/01/2025 trong mục Nâng cao (300), Amazon DynamoDB, Amazon EventBridge, Amazon Managed Service for Apache Flink, Amazon Rekognition, Amazon Simple Storage Service (S3), AWS Lambda, Storage, Hướng dẫn kỹ thuật 

      Tổ chức ở mọi quy mô đều phải đối mặt với một thách thức chung: quản lý, tổ chức và truy xuất hiệu quả khối lượng lớn nội dung kỹ thuật số. Từ hình ảnh, video đến tài liệu và dữ liệu ứng dụng, doanh nghiệp đang bị ngập trong thông tin cần được:

  • Lưu trữ an toàn
  • Truy cập nhanh chóng  
  • Phân tích hiệu quả

Khả năng trích xuất, quản lý và sử dụng metadata từ nội dung này là then chốt để:

  • Kích hoạt khả năng tìm kiếm mạnh mẽ
  • Duy trì quản trị dữ liệu
  • Thu được những insight có giá trị

      Tuy nhiên, khi khối lượng dữ liệu tăng theo cấp số nhân, các hệ thống tự xây dựng này gặp khó khăn trong việc mở rộng quy mô, thường trở nên phức tạp và cồng kềnh, đặc biệt khi chúng phải xử lý các cập nhật, bổ sung hoặc xóa objects và metadata liên quan thường xuyên. Việc chia sẻ và tích hợp metadata giữa các hệ thống khác nhau, cùng với việc theo dõi trạng thái object, càng tạo thêm gánh nặng không cần thiết. Với Amazon S3 Metadata và Amazon S3 Tables, bạn có thể giảm bớt việc tạo và tự động hóa quản lý metadata và trạng thái object trong một table bucket tương thích Apache Iceberg trên S3.

      Trong bài viết này, chúng tôi trình bày cách đơn giản hóa việc quản lý metadata bằng S3 Metadata và S3 Tables. Thông qua một ví dụ thực tế, tìm hiểu cách tích hợp S3 Metadata với custom metadata được tạo bởi Amazon Rekognition và công cụ Pillow, được lưu trữ trong table bucket S3 sử dụng ứng dụng Apache Flink.

Tổng quan về giải pháp

       Giải pháp sử dụng bảng tương thích Apache Iceberg trong S3 table bucket để lưu trữ content metadata của các S3 objects được tải lên bucket. Chúng tôi sử dụng S3 table này kết hợp với S3 Metadata table để trả lời các câu hỏi liên quan đến cả nội dung và trạng thái hiện tại của objects trong S3 bucket.

       Mẫu ứng dụng được cung cấp xây dựng một enriched event stream từ S3 Event Notifications được tạo ra bởi các thao tác PUT và DELETE vào bucket. Sau đó, nó commit các thay đổi trong stream đó vào bảng Apache Iceberg trong S3 table bucket. Giải pháp này sử dụng:

  • AWS Lambda để thực hiện việc trích xuất metadata từ nội dung của các object.
  • Amazon Rekognition để tạo labels.
  • Pillow để xác định kích thước hình ảnh (width và height).
  • Amazon DynamoDB table để đảm bảo thứ tự của các event.
  • Apache Flink để batch và commit các event vào bảng Iceberg trong S3 table bucket.

                                                  Hình 1: Tổng quan về giải pháp

Luồng dữ liệu qua hệ thống như sau:

  1. Images được upload lên bucket.
  1. S3 Metadata thu thập và ghi lại object metadata trong S3 table bucket.
  1. S3 Event Notifications được cấu hình để kích hoạt Lambda function thông qua Amazon EventBridge để thu thập custom metadata.
  1. Custom metadata về nội dung của object được tạo ra thông qua Amazon Rekognition và Pillow, sau đó được chèn vào DynamoDB table.
  1. Các enriched events được đọc từ DynamoDB Change Stream bởi Apache Flink.
  1. Apache Flink ghi records vào Apache Iceberg table được lưu trữ trong S3 table bucket.

       Khi thực hiện queries, bạn thường sử dụng object metadata được tạo bởi tính năng S3 Metadata hoặc custom content metadata được tạo bởi ứng dụng, hoặc kết hợp cả hai. Cả hai loại metadata đều được lưu trữ trong các bảng riêng biệt trong cùng một S3 table bucket.

Vai trò của DynamoDB và Apache Flink:

         S3 Event Notifications được thiết kế để gửi notifications ít nhất một lần, nhưng không đảm bảo chúng sẽ đến theo thứ tự các sự kiện xảy ra. Trong một số trường hợp hiếm gặp, cơ chế retry của S3 có thể gây ra trùng lặp S3 Event Notifications cho cùng một object event. Đôi khi việc xử lý event có thể mất thời gian khác nhau cho các mutation operations được thực hiện trên cùng một key. DynamoDB conditional-write capability ngăn chặn việc xử lý một event trước đó sau khi đã hoàn thành một event sau. Records được lưu trong DynamoDB chỉ trong 24 giờ trước khi hết hạn. Để biết thêm thông tin về event ordering và xử lý duplicate events, hãy đọc bài viết AWS Storage Blog có tựa đề “Manage event ordering and duplicate events with Amazon S3 Event Notifications”.

         Apache Iceberg là một open table format cho các big data analytics tables, thường được tạo thành từ nhiều Apache Parquet files. Các thao tác insert, update và deletion từ Iceberg table được thực hiện bằng cách đọc dữ liệu bảng hiện có, chèn, sửa đổi hoặc xóa rows từ data files, ghi ra các files mới với nội dung đã sửa đổi và commit các thay đổi vào metadata của bảng. Điều này có nghĩa là các thao tác insert hoặc modification ở mức row có thể cần đọc và ghi một lượng lớn dữ liệu trải rộng trên nhiều files. Do đó, Apache Flink được sử dụng trong ứng dụng mẫu này để gộp các modifications này thành bulk writes.

Thành phần giải pháp

The AWS Cloud Development Kit (AWS CDK) stack triển khai các tài nguyên sau:

  • Một input S3 bucket để tải lên tài nguyên mới.
  • Một Amazon EventBridge rule.
  • Lambda functions để trích xuất metadata từ hình ảnh.
  • Một DynamoDB table và change stream để giải quyết xung đột cập nhật đồng thời
  • Một Amazon Managed Service for Apache Flink application.
  • Một logical table và namespace trong S3 table bucket được cung cấp (được truyền dưới dạng tham số thông qua context) để lưu trữ custom metadata.
  • Roles và AWS Identity and Access Management (IAM) policies cần thiết để triển khai tài nguyên.

Điều kiện tiên quyết

Để triển khai AWS CDK stack, bạn cần các điều kiện tiên quyết sau:

  • Một tài khoản AWS
  • Phiên bản mới nhất của AWS Command Line Interface (AWS CLI) (tối thiểu 2.22.17) để hỗ trợ S3 Tables và S3 Metadata
  • Docker
  • Python 3.9 trở lên
  • NPM 10.7 trở lên
  • Apache Maven 3.2.5 trở lên
  • Java 11 trở lên
  • AWS CDK
  • IAM permissions để triển khai và sử dụng tài nguyên trong CDK stack cùng với S3 Tables và S3 Metadata
  • Để chạy các truy vấn, bạn cần cài đặt Apache Spark (3.5.2 trở lên) với Iceberg (1.6.0 trở lên) hoặc sử dụng Amazon EMR (7.5 trở lên).

Hướng dẫn chi tiết

      Trong hướng dẫn giải pháp này, bạn thực hiện các bước sau để thiết lập S3 Metadata và sử dụng nó với custom metadata.

  • Tạo một S3 table bucket.
  • Triển khai AWS CDK stack.
  • Cấu hình S3 Metadata.
  • Khởi động ứng dụng Managed Apache Flink.
  • Tải lên các hình ảnh mẫu vào input bucket.
  • Chạy các truy vấn để tích hợp custom metadata với Amazon S3 Metadata.

1. Tạo một S3 table bucket

      Cấu hình AWS CLI của bạn với AWS Region mặc định nơi S3 Metadata is available. Sau đó, tạo một S3 table bucket.

aws s3tables create-table-bucket –name mdblogtb –region us-east-1

       Lưu Amazon Resource Name (ARN) từ output, vì bạn sẽ sử dụng nó cho việc triển khai AWS CDK. Bạn cũng có thể sử dụng lệnh aws s3tables list-table-buckets –region us-east-1 để liệt kê các table bucket. Tên table bucket không phải là duy nhất toàn cầu, do đó bạn có thể tạo các table bucket với cùng tên ở các AWS Region khác nhau.

2. Triển khai AWS CDK stack

Mã nguồn mẫu có sẵn để tải về trên GitHub. Bắt đầu bằng cách clone repository.

git clone https://github.com/aws-samples/amazon-s3-contentmetadata.git

cd amazon-s3-contentmetadata

Chạy lệnh mvn package để biên dịch, xác minh và build các package cần thiết trong AWS CDK stack.

      Chạy các AWS CDK commands để bootstrap, synthesize, và deploy stack. Để tìm hiểu về các bước này, tham khảo AWS CDK documentation và tutorial. Bạn truyền context value cho S3 table bucket ARN, do đó AWS CDK có thể sử dụng nó để tạo Iceberg table mới để lưu trữ custom metadata.

Hình 2 thể hiện kết quả của lệnh CDK bootstrap.

                                                       Hình 2: AWS CDK bootstrap

Hình 3 hiển thị kết quả của lệnh CDK synth.

                                                       Hình 3: AWS CDK synth in progress

Deploy the stack:

cdk deploy –context s3_tables_bucket_arn=<ARN_TABLE_BUCKET>

Quá trình deployment xác định các thay đổi. Review các thay đổi và chấp nhận chúng bằng cách chọn y tại prompt, như hiển thị trong Hình 4.

                                                   Hình 4: AWS CDK deploy

     Sau khi hoàn thành thành công, bạn sẽ thấy output với S3 general purpose input bucket mới và Flink ApplicationName. Output mẫu được hiển thị trong Hình 5. Bạn có thể xem chi tiết stack deployment trong AWS CloudFormation service trong AWS Management Console. Input S3 bucket mới được sử dụng để upload objects cho custom metadata extraction..

                                                           Hình 5: AWS CDK output

Lưu output vào text editor, vì bạn sẽ sử dụng thông tin này trong các bước cấu hình tiếp theo.

3. Cấu hình S3 Metadata

     Tạo cấu hình S3 Metadata trên bucket general purpose mới theo documentation. Nhập ARN của table bucket mà bạn đã ghi lại trong quá trình tạo table bucket và sử dụng mdblogs3metadata làm tên bảng metadata. Cuối cùng, chọn Create metadata configuration, như được hiển thị trong Hình 6.

                                                 Hình 6: S3 Metadata configuration

     Trong output, lưu ý namespace aws_s3_metadata mới được tạo trong table bucket được cung cấp để lưu trữ metadata được tạo từ S3. Tên namespace này được tạo tự động bởi hệ thống và không thể thay đổi. Hãy ghi lại namespace và tên table, vì chúng sẽ được sử dụng trong các truy vấn.

4. Khởi động ứng dụng Managed Apache Flink

      Sử dụng tên ứng dụng được tạo từ output AWS CDK deploy, khởi động ứng dụng Apache Flink như sau, thay thế bằng tên ứng dụng thực tế của bạn. Ngoài ra, bạn có thể sử dụng list-applications `aws kinesisanalyticsv2 list-applications` để tìm ApplicationName nếu bạn không có output AWS CDK deploy.

aws kinesisanalyticsv2 start-application –application-name IcebergProcessor29058B24-85XGQ9V1ZqgV

       Bạn có thể truy cập vào dịch vụ Managed Apache Flink trong console để kiểm tra trạng thái. Có thể mất vài phút để ứng dụng khởi động, vì vậy bạn nên đợi cho đến khi trạng thái chuyển sang Running.

       Ở giai đoạn này, bạn có các table sau trong S3 table bucket mdblogtb của mình, với các mục đích được nêu như sau:

5. Upload sample images to input bucket

      Kho lưu trữ GitHub bao gồm các sample images và một script commands.sh để upload các objects vào content input bucket.

      Script tải lên các đối tượng với tags sử dụng key Project và các giá trị Africa hoặc Europe để mô phỏng các kịch bản nơi hình ảnh được chụp từ các lần sản xuất khác nhau. Mỗi lần sản xuất có thể có các loại nội dung khác nhau, ví dụ như hình ảnh từ rừng rậm có voi, động vật, địa điểm, trận đấu bóng đá, hoặc đồ nội thất. Amazon Rekognition nhận diện nội dung trong hình ảnh, và kích thước hình ảnh được trích xuất thông qua Pillow. Các tệp được tải lên được phân loại như sau:

      Kịch bản này phổ biến trong các trường hợp sử dụng lưu trữ media điển hình nhưng có thể được mở rộng và tùy chỉnh cho các dataset khác. Thông thường, khách hàng sử dụng các bucket hoặc prefix khác nhau để lưu trữ dữ liệu theo các danh mục khác nhau. Do đó, việc sử dụng tag là tùy chọn và chỉ nhằm mục đích demo.

6. Chạy các query để tích hợp custom metadata với S3 Metadata

      Để query các S3 Table, hãy làm theo hướng dẫn cho Amazon EMR. Ngoài ra, bạn có thể sử dụng Apache Spark được cài đặt trên server để chạy các Spark query. Bắt đầu phiên PySpark (điều chỉnh đường dẫn nếu cần) bằng cách thay thế <table_bucket_arn> bằng ARN chính xác của table bucket. Dưới đây là một lệnh mẫu:

Pyspark/opt/spark/bin/pyspark \–packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.0,software.amazon.s3tables:s3-tables-catalog-for-iceberg:0.1.3 \–conf “spark.driver.extraJavaOptions=-Djava.security.manager=allow” \–conf “spark.executor.extraJavaOptions=-Djava.security.manager=allow” \–conf “spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions” \–conf “spark.sql.catalog.s3tb=org.apache.iceberg.spark.SparkCatalog” \–conf “spark.sql.catalog.s3tb.catalog-impl=software.amazon.s3tables.iceberg.S3TablesCatalog” \–conf “spark.sql.catalog.s3tb.warehouse=<table_bucket_arn>” \–master “local[*]”

     Tiếp theo chạy các truy vấn để xác định các objects trong bucket dựa trên trạng thái của objects và object tags (sử dụng S3 Metadata table) và loại nội dung (sử dụng custom content metadata). Đầu tiên, chạy truy vấn sau để xem các mẫu bản ghi và schema của bảng trong cả hai bảng. Có thể mất vài phút để các bản ghi xuất hiện trong bảng mdblogs3metadata sau khi objects được tải lên.

    Kết quả mẫu của truy vấn được cung cấp trong Hình 7 để tham khảo.

                                                   Hình 7: Sample table output

Scenario #1

Ví dụ, bạn muốn xác định các objects, storage class của chúng và object tags liên quan đến Project Africa nơi có hình ảnh Elephant. Điều này có thể là do chiến dịch marketing cho buổi chụp này, hoặc có thể các hình ảnh cần được cấp phép, phân phối hoặc tái sử dụng cho một dự án khác.

                                                Hình 8: Output from first query

     Cả bảng S3 Metadata được tạo và bảng với thông tin nội dung tùy chỉnh đã được sử dụng thông qua một join.

Trong output, elephant-europe.jpg đã không được liệt kê, vì nó là một phần của Project Europe. Để tìm tất cả các objects có hình ảnh Elephant bất kể Project nào, hãy chạy truy vấn sau.

                                              Hình 9: Output from second query

Scenario #2

Để tìm các objects mà có Soccer Ball trong label, chạy query sau đây. Trong sample được cung cấp, những images này được gắn tag với Project Africa. Nếu có nhiều tags, bạn có thể sử dụng filter hoặc đơn giản là collect một list của các images trong bucket.

                                     Hình 10: Output from query without coalesced view

      Có nhiều mục nhập cho soccer.jpg. Để hiểu điều này, trước tiên hãy xem lại script commands.sh được sử dụng để upload objects. Bạn sẽ thấy chuỗi các thao tác object và record_types liên quan sau:

  • Object được upload lần đầu với tag sai Project=WrongProject. [record_type CREATE]
  • Object đã bị xóa, record này không hiển thị vì query khớp với eTag. Nếu bạn chạy query mà không có eTag match, bạn sẽ thấy record này. [record_type DELETE]
  • Object được upload lại với tag Project=Africa. [record_type CREATE và object_tags Project=Africa]
  • Object tags được sửa đổi bằng thao tác put-object-tagging, và các tag mới Object Info=Soccer Ball và Project=Africa được thêm vào. [record_type UPDATE_METADATA và object_tags Object Info=Soccer Ball và Project=Africa]

      Lý do có nhiều mục nhập cho soccer.jpg trong bảng S3 Metadata mdblogs3metadata là vì mỗi hàng đại diện cho một sự kiện mutation đã tạo, cập nhật hoặc xóa một object trong general purpose bucket của bạn. Các sự kiện trong bài này là kết quả của các hành động người dùng được mô phỏng bằng commands.sh, nhưng một số có thể là kết quả của các hành động được thực hiện bởi S3 thay mặt bạn, như S3 Lifecycle expirations hoặc lifecycle storage class transitions. S3 Metadata là một event-processing pipeline được thiết kế để giữ cho bảng metadata cuối cùng nhất quán với các thay đổi xảy ra trong general purpose bucket của bạn. Điều này cung cấp trạng thái của object, do đó loại bỏ nhu cầu về các hệ thống theo dõi phức tạp.

      Do đó, một coalesced view hiển thị trạng thái mới nhất của object. View có thể xác định phiên bản gần đây nhất của mỗi object bằng cách lọc ra các object đã xóa và đánh dấu phiên bản mới nhất của mỗi object dựa trên sequence numbers. Kết quả được sắp xếp theo các cột bucket, key và sequence_number. Tạo nó bằng lệnh sau:

Pysparkspark.sql(“”” \CREATE TEMPORARY VIEW s3_metadata_coalesced AS ( \WITH cte as ( \SELECT * from s3tb.aws_s3_metadata.mdblogs3metadata \), \version_stacks as ( \SELECT *, \LEAD(sequence_number, 1) over ( \partition by (bucket, key, version_id) order by sequence_number ASC \ ) as next_sequence_number \from cte \), \latest_versions as ( \SELECT * from version_stacks where next_sequence_number is NULL \), \extant_versions as ( \SELECT * from latest_versions where record_type != ‘DELETE’ \), \with_is_latest as ( \SELECT *, \sequence_number = (MAX(sequence_number) over (partition by (bucket, key))) as is_latest_version \FROM extant_versions \) \SELECT * from with_is_latest \ORDER BY bucket, key, sequence_number \) \”””).show()

Chúng tôi không tạo view cho custom metadata, vì bảng custom content metadata sẽ xóa bản ghi cho bất kỳ thao tác DELETE nào và thêm bản ghi cho các sự kiện PUT object. Bạn có thể xem rule EventBridge được triển khai bởi AWS CDK để thấy các loại sự kiện kích hoạt việc tạo custom content metadata thông qua một hàm Lambda.

Chúng ta có thể chạy lại truy vấn để tìm tất cả các hình ảnh có Soccer Ball, nhưng lần này chúng ta sử dụng view coalesced mới.

                                           Hình 11: Output from query with coalesced view

Lần này, chỉ có bản ghi mới nhất cho object soccer.jpg xuất hiện trong output, vì coalesced view có trạng thái được ghi nhận mới nhất của object là UPDATE_METADATA. Nếu thao tác cuối cùng là xóa, thì object sẽ không xuất hiện trong output, vì bản ghi tương ứng của nó trong custom content metadata table đã bị xóa.

Building workflow pipelines

Bạn có thể lưu trữ output của các query trong file comma-separated values (CSV) và sử dụng nó để xử lý thêm. Ví dụ:

Nếu bạn có các tiered objects trong storage class S3 Glacier Flexible Retrieval, bạn có thể lọc các bản ghi và sử dụng các key để gửi một S3 Batch Operations job để thực hiện restore.

Bạn có thể sử dụng S3 Batch copy job để copy object.

Bạn có thể trigger một Lambda để xử lý hình ảnh ở format/dimension mới hoặc copy sang bucket khác để chỉnh sửa.

Cleaning up

Clean up các tài nguyên đã triển khai để tránh các chi phí trong tương lai. Thực hiện các bước sau:

Empty S3 bucket đã được sử dụng cho input/image uploads.

Đi đến thư mục chính nơi bạn đã clone GitHub repository và chạy `CDK destroy`.

Kết luận

Trong bài viết này, chúng tôi đã trình bày một giải pháp mẫu để trích xuất custom metadata từ các objects bằng Amazon Rekognition và Pillow. Dữ liệu này được lưu trữ trong S3 table bucket. Chúng tôi cũng sử dụng S3 Metadata để trích xuất object metadata, cũng được lưu trữ trong S3 table bucket. Hơn nữa, chúng tôi đã trình bày cách sử dụng cả hai bảng để chạy queries và trích xuất thông tin về objects dựa trên các content types nhất định. Chúng tôi cũng chia sẻ cách tạo coalesced view của bảng S3 Metadata để xem trạng thái mới nhất của objects và sử dụng nó trong các queries.

Như một bước tiếp theo, chúng tôi khuyến nghị bạn xem xét sample code của chúng tôi và sử dụng nó như nguồn cảm hứng để đáp ứng nhu cầu kinh doanh của riêng bạn. Bạn có thể tùy chỉnh giải pháp bằng cách sử dụng Amazon Bedrock hoặc custom content extraction tool của bạn, hoặc thậm chí chia sẻ/xuất dữ liệu từ content management systems hoặc digital asset management solutions hiện có của bạn vào S3 Tables để index, join và tìm kiếm objects và content metadata của chúng.

Nhìn chung, S3 Metadata và S3 Tables giúp tự động hóa và loại bỏ undifferentiated heavy lifting cần thiết để duy trì trạng thái của objects và metadata liên quan của chúng. Hãy bắt đầu xây dựng!

Cảm ơn bạn đã đọc bài viết này. Nếu bạn có bất kỳ bình luận hoặc câu hỏi nào, hãy để lại trong phần bình luận.

Bài được dịch từ bài viết trên AWS Blogs, bạn có thể xem bài viết gốc tại đây.

←
→

Leave a comment