AWS Study Group

Truyền dữ liệu từ Amazon MSK đến bảng Apache Iceberg trên Amazon S3 và Amazon S3 Tables bằng Amazon Data Firehose

Tác giả: Pratik Patel, Amar Surjit, và Priyanka Chaudhary

Ngày xuất bản: ngày 20 tháng 06 năm 2025

Danh mục:  Amazon Data Firehose, Amazon S3 Tables, Amazon Simple Storage Service (S3), Architecture, AWS Big Data

Trong bối cảnh kinh doanh hiện đại, nơi dữ liệu đóng vai trò then chốt và nhịp độ thay đổi cực nhanh, phân tích dữ liệu streaming thời gian thực (real-time streaming analytics) đã trở thành yếu tố sống còn cho thành công của doanh nghiệp. Từ việc phát hiện các giao dịch gian lận trong lĩnh vực tài chính, theo dõi dữ liệu cảm biến từ các thiết bị Internet vạn vật (IoT) trong sản xuất, cho đến phân tích hành vi người dùng trên các nền tảng thương mại điện tử, thì việc phân tích dữ liệu theo thời gian thực đều giúp các tổ chức đưa ra quyết định nhanh chóng và phản ứng kịp thời trước các cơ hội cũng như rủi ro khi chúng vừa xuất hiện.

Trước nhu cầu ngày càng tăng, nhiều tổ chức đã bắt đầu áp dụng Apache Iceberg – một định dạng bảng mã nguồn mở hiện đại giúp đơn giản hóa việc xử lý các tập dữ liệu lớn được lưu trữ trong data lake. Iceberg mang lại trải nghiệm quen thuộc như SQL khi làm việc với dữ liệu lớn, đồng thời cung cấp nhiều khả năng mạnh mẽ như: giao dịch ACID, thao tác cấp độ dòng (row-level operations), tiến hóa phân vùng (partition evolution), phiên bản hóa dữ liệu (data versioning), xử lý gia tăng (incremental processing), và quét truy vấn nâng cao. Iceberg tích hợp liền mạch với các framework xử lý dữ liệu mã nguồn mở phổ biến như Apache Spark, Apache Hive, Apache Flink, Presto và Trino. Amazon Simple Storage Service (Amazon S3) hỗ trợ các bảng Iceberg theo cả hai cách: sử dụng trực tiếp định dạng bảng Iceberg và thông qua Amazon S3 Tables

Mặc dù Amazon Managed Streaming for Apache Kafka (Amazon MSK) cung cấp các khả năng streaming mạnh mẽ, có thể mở rộng cho nhu cầu dữ liệu thời gian thực, nhiều khách hàng cần một phương pháp hiệu quả và liền mạch để đưa dữ liệu streaming của họ từ Amazon MSK đến các bảng Iceberg trong Amazon S3 và S3 Tables. Đây chính là lúc Amazon Data Firehose (Firehose) phát huy tác dụng. Với sự hỗ trợ tích hợp sẵn cho các bảng Iceberg trong Amazon S3 và S3 Tables, Firehose giúp bạn có thể liền mạch đưa dữ liệu streaming từ các cụm MSK được cấp phát (provisioned clusters) đến các bảng Iceberg trong Amazon S3 và S3 Tables.

Là một dịch vụ trích xuất, chuyển đổi và tải (ETL) được quản lý hoàn toàn, Firehose đọc dữ liệu từ các topic Apache Kafka của bạn, chuyển đổi các bản ghi và ghi chúng trực tiếp vào các bảng Iceberg trong data lake của bạn trên Amazon S3. Khả năng mới này không đòi hỏi bạn phải viết mã hay quản lý cơ sở hạ tầng, cho phép tải dữ liệu liên tục và hiệu quả từ Amazon MSK vào Iceberg trên Amazon S3. Trong bài viết này, chúng tôi sẽ hướng dẫn hai giải pháp minh họa cách truyền dữ liệu từ cụm Amazon MSK được cấp phát của bạn đến các data lake dựa trên Iceberg trong Amazon S3 bằng Firehose.

Tổng quan về giải pháp 1: Bảng Amazon MSK tới Iceberg trong Amazon S3

Sơ đồ sau minh họa kiến trúc tổng quan dạng cấp cao trong việc phân phối dữ liệu streaming từ Amazon MSK đến các bảng Iceberg được lưu trữ trong Amazon S3.

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

Để có thể thực hiện các bước theo hướng dẫn trong bài viết này, bạn cần đáp ứng các điều kiện tiên quyết sau:

  • Có một tài khoản AWS
  • Một S3 bucket
  • Một bảng Iceberg đã được đăng ký trong AWS Glue Data Catalog
  • Một cụm Amazon MSK (được cấp phát – provisioned) đang hoạt động, có bật xác thực kiểm soát truy cập bằng AWS Identity and Access Management (IAM) và hỗ trợ kết nối đa VPC (multi-VPC connectivity). Tham khảo phần Configure source settings for Amazon MSK để biết hướng dẫn bật hỗ trợ kết nối đa VPC cho Amazon MSK Cluster.

Xác minh quyền truy cập

Trước khi cấu hình luồng phân phối dữ liệu (Firehose delivery stream), bạn cần xác minh rằng bảng đích đã tồn tại trong Data Catalog.

  1. Trong giao diện AWS Glue, điều hướng đến mục Glue Data Catalog và xác nhận rằng bảng Iceberg đã tồn tại với đầy đủ các thuộc tính cần thiết
  1. Xác nhận rằng cụm Amazon MSK (được cấp phát) của bạn đang hoạt động, đã bật xác thực IAM và hỗ trợ kết nối đa VPC (multi-VPC connectivity)
  1. Cấp cho Firehose quyền truy cập cụm MSK riêng tư của bạn:
    1. Trong giao diện Amazon MSK, đi đến mục cluster và chọn Properties and Security settings.
    2. Chỉnh sửa chính sách của cụm và định nghĩa một chính sách tương tự như bên dưới: 
{

  "Version": "2012-10-17",

  "Statement": [

    {

      "Principal": {

        "Service": [

          "firehose.amazonaws.com"

        ]

    },

    "Effect": "Allow",

    "Action": [

      "kafka:CreateVpcConnection"

    ],

    "Resource": "<Amazon MSK cluster-arn>"

    }

  ]

}

Điều này đảm bảo rằng Firehose có đầy đủ quyền cần thiết trên cụm Amazon MSK nguồn đã được cấp phát. 

Tạo role cho Firehose

Phần này mô tả các quyền cho phép Firehose nhập liệu (ingest), xử lý và phân phối dữ liệu từ nguồn đến đích. Bạn phải chỉ định một IAM role cấp quyền cho Firehose để nhập dữ liệu nguồn từ cụm Amazon MSK được cấp phát. Hãy đảm bảo rằng các chính sách tin cậy (trust policy) sau được đính kèm vào role đó để Firehose có thể đảm nhận vai trò (assume role):

{

  "Version": "2012-10-17",

  "Statement": [

    {

      "Principal": {

        "Service": [

          "firehose.amazonaws.com"

        ]

      },

      "Effect": "Allow",

      "Action": "sts:AssumeRole"

    }

  ]

}

Hãy đảm bảo rằng role IAM này cấp cho Firehose các quyền cần thiết để thu thập dữ liệu nguồn từ cụm Amazon MSK (được cấp phát) đã chỉ định:

{

   "Version": "2012-10-17",      

   "Statement": [{

        "Effect":"Allow",

        "Action": [

           "kafka:GetBootstrapBrokers",

           "kafka:DescribeCluster",

           "kafka:DescribeClusterV2",

           "kafka-cluster:Connect"

         ],

         "Resource": "<CLUSTER-ARN>"

       },

       {

         "Effect":"Allow",

         "Action": [

           "kafka-cluster:DescribeTopic",

           "kafka-cluster:DescribeTopicDynamicConfiguration",

           "kafka-cluster:ReadData"

         ],

         "Resource": "<TOPIC-ARN>"

       }]

}

Hãy đảm bảo rằng vai trò Firehose có quyền truy cập vào Glue Data Catalog và S3 bucket:

{

    "Version": "2012-10-17",  

    "Statement":

    [    

        {      

            "Effect": "Allow",      

            "Action": [

                "glue:GetTable",

                "glue:GetDatabase",

                "glue:UpdateTable"

            ],      

            "Resource": [   

                "arn:aws:glue:<region>:<aws-account-id>:catalog",

                "arn:aws:glue:<region>:<aws-account-id>:database/*",

                "arn:aws:glue:<region>:<aws-account-id>:table/*/*"             

            ]    

        },        

        {      

            "Effect": "Allow",      

            "Action": [

                "s3:AbortMultipartUpload",

                "s3:GetBucketLocation",

                "s3:GetObject",

                "s3:ListBucket",

                "s3:ListBucketMultipartUploads",

                "s3:PutObject",

                "s3:DeleteObject"

            ],      

            "Resource": [   

                "arn:aws:s3:::<S3 bucket name>",

                "arn:aws:s3:::<S3 bucket name>/*"              

            ]    

        } 

    ]

}

Tham khảo các chính sách chi tiết hơn ở các nguồn sau: 

Bây giờ, khi bạn đã xác nhận rằng cụm MSK nguồn và bảng Iceberg đích đều đã sẵn sàng, bạn có thể bắt đầu cấu hình Firehose để chuyển dữ liệu streaming đến các bảng Iceberg trong Amazon S3.

Tạo một luồng Firehose

Hoàn thành các bước sau để tạo một luồng Firehose: 

  1. Trong giao diện Firehose, chọn Create Firehose stream.
  2. Chọn Amazon MSK cho Source và Apache Iceberg Tables cho Destination.
  1. Đặt tên cho luồng Firehose và chỉ định cấu hình cụm.
  1. Bạn có thể chọn một cụm MSK ở tài khoản hiện tại hoặc ở một tài khoản khác. 
  2. Để có thể chọn cụm, cụm đó bắt buộc phải đang ở trạng thái hoạt động (active), sử dụng IAM như một trong các phương thức kiểm soát truy cập, và đã bật kết nối đa VPC (multi-VPC connectivity).
  1. Cung cấp tên MSK topic mà Firehose sẽ đọc dữ liệu.
  1. Điền tên Firehose stream
  1. Tùy chỉnh các thiết lập cho điểm đích – destination dữ liệu sẽ đến, nơi bạn có thể chọn gửi dữ liệu trong tài khoản hiện tại hoặc sang tài khoản khác.
  2. Chọn vị trí tài khoản là tài khoản hiện tại (Current account), chọn vùng AWS (AWS Region) phù hợp, và trong phần Catalog, hãy chọn ID tài khoản hiện tại.

Để định tuyến dữ liệu streaming đến các bảng Iceberg khác nhau và thực hiện các hoạt động như chèn (insert), cập nhật (update) và xóa (delete), bạn có thể sử dụng các biểu thức JQ của Firehose. Bạn có thể tìm thấy thông tin cần thiết tại đây.

  1. Cung cấp cấu hình khóa chính (unique key), khóa này sẽ được sử dụng để thực hiện các thao tác cập nhật và xóa dữ liệu.
  1. Đến phần Buffer hints và cấu hình kích thước bộ đệm (Buffer size) là 1 MiB và khoảng thời gian đệm (Buffer interval) là 60 giây.  Bạn có thể điều chỉnh các thiết lập này tùy theo nhu cầu của trường hợp sử dụng cụ thể.
  2. Cấu hình thiết lập sao lưu (backup) bằng cách chỉ định một S3 backup bucket.

Với Firehose, bạn có thể cấu hình sao lưu bằng cách cung cấp một bucket S3 dùng cho sao lưu, kèm theo các tiền tố tùy chỉnh như error , để các bản ghi thất bại được tự động lưu lại và có thể truy xuất để xử lý sự cố hoặc chạy lại sau này.

  1. Trong phần Cấu hình nâng cao (Advanced settings), hãy bật tính năng ghi log lỗi bằng Amazon CloudWatch.
  1. Trong phần Service access, chọn IAM role mà bạn đã tạo trước đó cho Firehose. 
  2. Xem lại các thiết lập của bạn và chọn Create Firehose stream.

Luồng Firehose sẽ sẵn sàng và bắt đầu truyền dữ liệu từ các chủ đề (topic) trong MSK đến bảng Iceberg trong Amazon S3.

Bạn có thể truy vấn dữ liệu trong bảng bằng Amazon Athena để kiểm tra dữ liệu streaming

  1. Trong giao diện Athena console, mở trình soạn thảo truy vấn (query editor).
  2. Chọn bảng Iceberg và chạy xem trước bảng (table preview).

Lúc này, bạn sẽ có thể truy cập dữ liệu streaming đã được ghi vào bảng.

Tổng quan giải pháp 2: Từ Amazon MSK đến Amazon S3 Tables

S3 Tables được phát triển dựa trên định dạng bảng mã nguồn mở của Iceberg, giúp bạn quản lý dữ liệu trong Amazon S3 giống như làm việc với bảng trong cơ sở dữ liệu. Bạn có thể tổ chức và truy vấn dữ liệu bằng cách sử dụng các cú pháp quen thuộc dùng cho bảng, đồng thời tận dụng các tính năng mạnh mẽ của Iceberg như tiến hóa lược đồ (schema evolution), tiến hóa phân vùng (partition evolution), và truy xuất lịch sử (time travel) ngay trên Amazon S3. Tính năng này hỗ trợ các giao dịch tuân thủ theo quy tắc ACID trong dữ liệu, và cho phép thực hiện các thao tác INSERT, UPDATE và DELETE trên dữ liệu trong Amazon S3, giúp việc quản lý data lake trở nên hiệu quả và đáng tin cậy hơn.

Bạn có thể sử dụng Firehose để truyền dữ liệu streaming từ một cụm Amazon MSK (được cấp phát) đến các bảng Iceberg trong Amazon S3. Bằng cách tạo một S3 table bucket trong  giao diện Amazon S3 Console, và bucket này sẽ được đăng ký với AWS Lake Formation, cho phép bạn quản lý quyền truy cập chi tiết (fine-grained access control) cho hệ thống data lake dựa trên Iceberg của mình trên S3 Tables.

Sơ đồ sau minh họa kiến trúc giải pháp này.

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

Trước khi bắt đầu, bạn cần đáp ứng những điều kiện tiên quyết sau:

  • Một tài khoản AWS
  • Một cụm Amazon MSK (đã được cấp phát) được khởi tạo và chạy, và xác thực truy cập bằng IAM đã được bật, đồng thời cụm hỗ trợ kết nối đa VPC (multi-VPC connectivity)
  • Một vai trò IAM dành cho Firehose (đã tạo ở bước trước), kèm theo chính sách quyền (IAM policy) bổ sung sau:
{

    "Version": "2012-10-17",

    "Statement": [

        {

            "Sid": "Statement1",

            "Effect": "Allow",

            "Action": [

                "lakeformation:GetDataAccess"

            ],

            "Resource": [

                "*"

            ]

        }

    ]

}

Tiếp theo, trong mục role IAM của Firehose, hãy thêm s3tablescatalog vào mục tài nguyên (resource) để cấp quyền truy cập vào S3 Table, như hình minh họa bên dưới.

Tạo một S3 table bucket

Tham khảo Creating a table bucket để tạo một S3 table bucket trong giao diện Amazon S3.

Khi bạn tạo bucket bảng đầu tiên với tùy chọn Cho phép tích hợp (Enable integration), Amazon S3 sẽ tự động cố gắng tích hợp bucket đó với các dịch vụ phân tích dữ liệu của AWS. Việc tích hợp này giúp bạn có thể sử dụng các dịch vụ phân tích của AWS để truy vấn toàn bộ các bảng dữ liệu (tables) trong cùng một khu vực (Region). Đây là bước quan trọng trong quá trình thiết lập tiếp theo. Nếu việc tích hợp này đã được thực hiện từ trước, bạn có thể sử dụng AWS Command Line Interface (AWS CLI) theo cách sau:

aws s3tables create-table-bucket –region <region id> –name <bucket name>

Tạo một namespace

Namespace của S3 Tables là một cấu trúc logic trong một table bucket. Mỗi bảng thuộc về một không gian tên (namespace) đơn lẻ. Trước khi tạo một bảng, bạn buộc phải tạo một namespace để nhóm các bảng lại với nhau. Bạn có thể tạo không gian tên bằng cách sử dụng Amazon S3 REST API, AWS SDK, AWS CLI, hoặc các công cụ truy vấn tích hợp.

Bạn có thể sử dụng lệnh AWS CLI sau để tạo một không gian tên bảng (table namespace):

aws s3tables create-namespace --table-bucket-arn

arn:aws:s3tables:us-east-1:111122223333:bucket/amzn-s3-demo-bucket --namespace example_namespace

Tạo một table

Một bảng S3 (S3 table) là một tài nguyên con (sub-resource) của một bucket bảng (table bucket). Tài nguyên này lưu trữ các bảng theo định dạng Iceberg, cho phép bạn làm việc với chúng thông qua các công cụ truy vấn và ứng dụng khác hỗ trợ định dạng Iceberg. Bạn có thể tạo một bảng bằng cách sử dụng lệnh AWS CLI sau:

aws s3tables create-table –cli-input-json file://mytabledefinition.json

Đoạn mã sau đây là nội dung của tệp mytabledefinition.json:

{

    "tableBucketARN": "arn:aws:s3tables:us-east-1:111122223333:bucket/amzn-s3-demo-table-bucket",

    "namespace": "example_namespace ",

    "name": "example_table",

    "format": "ICEBERG",

    "metadata": {

        "iceberg": {

            "schema": {

                "fields": [

                     {"name": "id", "type": "int", "required": true},

                     {"name": "name", "type": "string"},

                     {"name": "value", "type": "int"}

                ]

            }

        }

    }

}

Giờ đây, bạn đã có bảng cần thiết với đầy đủ các thuộc tính liên quan để sử dụng trong Lake Formation.

Cấp quyền Lake Formation cho tài nguyên bảng của bạn

Sau khi tích hợp, Lake Formation sẽ quản lý các quyền truy cập vào các tài nguyên bảng của bạn. Lake Formation sử dụng mô hình cấp quyền riêng, cho phép kiểm soát truy cập chi tiết đến từng bảng trong Glue Data Catalog. Để cho phép Firehose ghi dữ liệu vào S3 Tables, bạn có thể cấp quyền Lake Formation cho một principal (trong trường hợp này là IAM role của Firehose) trên một bảng trong table bucket, thông qua Lake Formation console hoặc AWS CLI.:

  1. Đảm bảo bạn đang chạy các lệnh AWS CLI với quyền quản trị viên của data lake. Xem hướng dẫn tại: Create a data lake administrator.
  2. Chạy lệnh sau để cấp quyền Lake Formation trên bảng trong bucket bảng S3 cho IAM principal (vai trò Firehose):
aws lakeformation grant-permissions \

--region <region e.g. us-east-1> \

--cli-input-json \

'{

    "Principal": {

        "DataLakePrincipalIdentifier": "<Amazon Data Firehose role ARN e.g. arn:aws:iam::<accound-id>:role/ExampleRole>"

    },

    "Resource": {

        "Table": {

            "CatalogId": "<account-id>:<s3tablescatalog>/<S3 table bucket name>",

            "DatabaseName": "<S3 table bucket namespace e.g. test_namespace>",

            "Name": "<S3 table bucket table name e.g. test_table>"

        }

    },

    "Permissions": [

        "ALL"

    ]

}'

Thiết lập một luồng Firehose ghi dữ liệu đến bảng S3 

Để thiết lập một luồng Firehose ghi dữ liệu vào bảng S3 sử dụng định dạng Iceberg thông qua bảng điều khiển Firehose, hãy thực hiện các bước sau:

  1. Ở giao diện Firehose console, chọn Create Firehose stream.
  2. Phần Source, chọn Amazon MSK.
  3. Ở phần Destination, chọn Apache Iceberg Tables.
  4. Đặt tên cho luồng Firehose.
  5. Cấu hình nguồn dữ liệu của bạn.
  6. Tại phần Destination settings, chọn Current Account, sau đó chọn khu vực của bạn (Region), và nhập tên table bucket mà bạn muốn ghi dữ liệu vào.
  7. Cấu hình tên database và bảng thông qua một trong các cách sau: sử dụng Unique Key configuration settings,  dùng biểu thức JSONQuery,  hoặc dùng một hàm AWS Lambda.

Xem thêm tại: Route incoming records to a single Iceberg table và Route incoming records to different Iceberg tables. 

  1. Phần Backup settings, chỉ định bucket S3 dùng để sao lưu bản ghi gốc.
  2. Ở phần Existing IAM roles của Advanced settings, chọn IAM role đã được cấu hình để Firehose có quyền ghi vào bảng S3.
  3. Chọn Create Firehose stream.

Sau khi hoàn tất, luồng Firehose sẽ bắt đầu truyền dữ liệu từ topic trong Amazon MSK vào bảng Iceberg trên S3. Bạn có thể xác minh bằng cách chạy truy vấn Athena trên bảng Iceberg để kiểm tra dữ liệu đã được ghi thành công.

Dọn dẹp tài nguyên

Tốt nhất nên dọn dẹp các tài nguyên đã tạo trong quá trình thực hiện hướng dẫn này, giúp tránh phát sinh chi phí không cần thiết. Để dọn dẹp, bạn hãy xóa các tài nguyên sau: cụm MSK, luồng Firehose đã tạo, bucket bảng S3 định dạng Iceberg, bucket S3 thông thường, và logs của CloudWatch.

Kết luận

Trong bài viết này, chúng tôi đã minh họa hai phương pháp để truyền dữ liệu theo thời gian thực từ Amazon MSK vào data lake bằng Firehose: truyền trực tiếp vào bảng Iceberg trên Amazon S3 và cách truyền vào S3 Tables. Firehose giúp đơn giản hóa quá trình quản lý pipeline dữ liệu truyền thống bằng cách cung cấp một giải pháp hoàn toàn do AWS quản lý (fully managed), phương pháp không cần viết mã (no-code), đồng thời tự động xử lý các tác vụ như chuyển đổi dữ liệu, nén và xử lý lỗi. Việc tích hợp liền mạch giữa Amazon MSK, Firehose và định dạng Iceberg trên S3 thể hiện cam kết của AWS trong việc đơn giản hóa kiến trúc dữ liệu lớn (big data) mà vẫn đảm bảo đầy đủ các tính năng như tuân thủ quy tắc ACID và khả năng truy vấn nâng cao, vốn là yêu cầu thiết yếu của các hệ thống data lake hiện đại. Chúng tôi hy vọng bài viết này hữu ích và khuyến khích bạn thử áp dụng giải pháp này để đơn giản hóa quy trình xử lý dữ liệu streaming vào các bảng Iceberg.

Về các tác giả

Pratik Patel

Pratik Patel là Kỹ sư quản lý tài khoản kỹ thuật cao cấp (Sr. Technical Account Manager) và chuyên gia trong lĩnh vực phân tích dữ liệu thời gian thực. Anh làm việc trực tiếp với khách hàng của AWS, cung cấp hỗ trợ liên tục và tư vấn kỹ thuật nhằm giúp họ lập kế hoạch, xây dựng giải pháp theo các phương pháp tốt nhất, đồng thời chủ động duy trì môi trường AWS luôn ổn định và hiệu quả.

Amar

Amar là một chuyên gia phân tích dữ liệu giàu kinh nghiệm tại AWS Vương quốc Anh. Anh hỗ trợ khách hàng của AWS triển khai các giải pháp dữ liệu quy mô lớn, với chuyên môn sâu về các dịch vụ phân tích và máy học của AWS. Amar giúp các tổ chức chuyển đổi theo hướng dữ liệu (data-driven transformation) và thúc đẩy đổi mới sáng tạo. Anh đặc biệt đam mê xây dựng các giải pháp có tác động lớn, và thường xuyên tham gia vào cộng đồng công nghệ để chia sẻ kiến thức và các phương pháp hay nhất trong lĩnh vực phân tích dữ liệu.

Priyanka Chaudhary

Priyanka Chaudhary là Kiến trúc sư giải pháp cấp cao (Senior Solutions Architect) và chuyên gia phân tích dữ liệu. Cô là một cố vấn đáng tin cậy cho khách hàng AWS, cung cấp định hướng kỹ thuật và hỗ trợ trong việc xây dựng các giải pháp theo tiêu chuẩn Well-Architected, sáng tạo và phù hợp với từng ngành nghề.

←
→

Leave a comment