Quy trình công việc được quản lý của Amazon cho Luồng khí Apache (Amazon MWAA) là dịch vụ điều phối được quản lý dành cho Luồng khí Apache điều đó giúp việc thiết lập và vận hành đường ống dữ liệu từ đầu đến cuối trên đám mây trở nên đơn giản.

Các tổ chức sử dụng Amazon MWAA để nâng cao quy trình công việc kinh doanh của họ. Ví dụ, Bộ gen C2i sử dụng Amazon MWAA trong nền tảng dữ liệu của họ để sắp xếp việc xác thực các thuật toán xử lý dữ liệu gen ung thư trong hàng tỷ bản ghi. Twitch, một nền tảng phát trực tiếp, quản lý và điều phối việc đào tạo cũng như triển khai các mô hình đề xuất của mình cho hơn 140 triệu người dùng đang hoạt động. Họ sử dụng Amazon MWAA để mở rộng quy mô, đồng thời cải thiện đáng kể tính bảo mật và giảm chi phí quản lý cơ sở hạ tầng.

Hôm nay, chúng tôi công bố sự sẵn có của môi trường Apache Airflow phiên bản 2.8.1 trên Amazon MWAA. Trong bài đăng này, chúng tôi sẽ hướng dẫn bạn một số tính năng và khả năng mới của Airflow hiện có trên Amazon MWAA cũng như cách bạn có thể thiết lập hoặc nâng cấp môi trường Amazon MWAA của mình lên phiên bản 2.8.1.

Lưu trữ đối tượng

Khi quy mô đường ống dữ liệu, các kỹ sư gặp khó khăn trong việc quản lý bộ nhớ trên nhiều hệ thống bằng các API, phương thức xác thực và quy ước duy nhất để truy cập dữ liệu, yêu cầu logic tùy chỉnh và toán tử dành riêng cho bộ nhớ. Airflow hiện cung cấp lớp trừu tượng lưu trữ đối tượng thống nhất để xử lý các chi tiết này, cho phép các kỹ sư tập trung vào đường dẫn dữ liệu của họ. Sử dụng lưu trữ đối tượng luồng không khí fsspec để cho phép mã truy cập dữ liệu nhất quán trên các hệ thống lưu trữ đối tượng khác nhau, từ đó hợp lý hóa độ phức tạp của cơ sở hạ tầng.

Sau đây là một số lợi ích chính của tính năng này:

• Quy trình làm việc di động – Bạn có thể chuyển đổi các dịch vụ lưu trữ với những thay đổi tối thiểu trong Đồ thị chu kỳ có hướng (DAG)
• Truyền dữ liệu hiệu quả – Bạn có thể truyền dữ liệu thay vì tải vào bộ nhớ
• Giảm bảo trì – Bạn không cần các toán tử riêng biệt, giúp cho việc bảo trì đường ống của bạn trở nên dễ dàng
• Kinh nghiệm lập trình quen thuộc – Bạn có thể sử dụng các mô-đun Python, như đóng cửa, cho các thao tác với tập tin

Để sử dụng bộ nhớ đối tượng với Dịch vụ lưu trữ đơn giản của Amazon (Amazon S3), bạn cần phải cài đặt thêm gói s3fs với nhà cung cấp Amazon (apache-airflow-providers-amazon[s3fs]==x.x.x).

Trong mã mẫu bên dưới, bạn có thể xem cách di chuyển dữ liệu trực tiếp từ Lưu trữ đám mây của Google tới Amazon S3. Bởi vì bộ lưu trữ đối tượng của Airflow sử dụng shutil.copyfileobj, dữ liệu của đối tượng được đọc theo từng đoạn từ gcs_data_source và truyền tới amazon_s3_data_target.

gcs_data_source = ObjectStoragePath("gcs://source-bucket/prefix/", conn_id="google_cloud_default")

amazon_s3_data_target = ObjectStoragePath("s3://target-bucket/prefix/", conn_id="aws_default ")

with DAG(
    dag_id="copy_from_gcs_to_amazon_s3",
    start_date=datetime(2024, 2, 26),
    schedule="0 0 * * *",
    catchup=False,    
    tags=["2.8", "ObjectStorage"],
) as dag:

    def list_objects(path: ObjectStoragePath) -> list[ObjectStoragePath]:
        objects = [f for f in path.iterdir() if f.is_file()]
        return objects

    def copy_object(path: ObjectStoragePath, object: ObjectStoragePath):    
        object.copy(dst=path)

    objects_list = list_objects(path=gcs_data_source)
    copy_object.partial(path=amazon_s3_data_target).expand(object=objects_list)

Để biết thêm thông tin về lưu trữ đối tượng Airflow, hãy tham khảo Lưu trữ đối tượng.

Giao diện người dùng XCom

XCom (giao tiếp chéo) cho phép truyền dữ liệu giữa các tác vụ, tạo điều kiện thuận lợi cho việc liên lạc và phối hợp giữa chúng. Trước đây, các nhà phát triển phải chuyển sang chế độ xem khác để xem XCom liên quan đến một nhiệm vụ. Với Airflow 2.8, các khóa-giá trị XCom được hiển thị trực tiếp trên một tab trong chế độ xem Airflow Grid, như được hiển thị trong ảnh chụp màn hình sau.

Mới xcom tab cung cấp các lợi ích sau:

• Cải thiện khả năng hiển thị XCom – Một tab chuyên dụng trong giao diện người dùng cung cấp một cách thuận tiện và thân thiện với người dùng để xem tất cả các XCom được liên kết với DAG hoặc nhiệm vụ.
• Cải thiện gỡ lỗi – Việc có thể xem trực tiếp các giá trị XCom trong giao diện người dùng rất hữu ích cho việc gỡ lỗi DAG. Bạn có thể nhanh chóng xem đầu ra của các tác vụ ngược dòng mà không cần phải kéo và kiểm tra chúng theo cách thủ công bằng mã Python.

Trình ghi bối cảnh tác vụ

Quản lý vòng đời tác vụ là rất quan trọng để đường ống dữ liệu trong Airflow vận hành trơn tru. Tuy nhiên, vẫn tồn tại một số thách thức nhất định, đặc biệt là trong các tình huống mà nhiệm vụ bị dừng đột ngột. Điều này có thể xảy ra do nhiều lý do, bao gồm cả thời gian chờ của bộ lập lịch, zombie nhiệm vụ (tác vụ vẫn ở trạng thái chạy mà không gửi nhịp tim) hoặc trường hợp nhân viên hết bộ nhớ.

Theo truyền thống, những lỗi như vậy, đặc biệt là những lỗi do các thành phần Airflow cốt lõi như bộ lập lịch hoặc bộ thực thi gây ra, không được ghi lại trong nhật ký tác vụ. Hạn chế này yêu cầu người dùng phải khắc phục sự cố bên ngoài giao diện người dùng Airflow, làm phức tạp quá trình xác định và giải quyết vấn đề.

Airflow 2.8 đã giới thiệu một cải tiến đáng kể nhằm giải quyết vấn đề này. Các thành phần luồng khí, bao gồm bộ lập lịch và bộ thực thi, giờ đây có thể sử dụng giao diện mới TaskContextLogger để chuyển tiếp thông báo lỗi trực tiếp đến nhật ký tác vụ. Tính năng này cho phép bạn xem tất cả các thông báo lỗi liên quan đến việc chạy một tác vụ ở một nơi. Điều này giúp đơn giản hóa quá trình tìm hiểu lý do tại sao một nhiệm vụ không thành công, cung cấp góc nhìn đầy đủ về những gì đã xảy ra trong một chế độ xem nhật ký.

Ảnh chụp màn hình sau đây cho thấy cách tác vụ được phát hiện dưới dạng zombievà nhật ký người lập lịch trình đang được đưa vào như một phần của nhật ký tác vụ.

Bạn cần đặt tham số cấu hình môi trường enable_task_context_logger đến True, để kích hoạt tính năng này. Sau khi được bật, Airflow có thể gửi nhật ký từ trình lập lịch trình, trình thực thi hoặc bối cảnh chạy lệnh gọi lại đến nhật ký tác vụ và cung cấp chúng trong giao diện người dùng Airflow.

Móc nghe cho tập dữ liệu

Bộ dữ liệu đã được giới thiệu trong Airflow 2.4 dưới dạng một nhóm nguồn dữ liệu hợp lý để tạo lập kế hoạch nhận biết dữ liệu và sự phụ thuộc giữa các DAG. Ví dụ: bạn có thể lên lịch chạy DAG tiêu dùng khi DAG sản xuất cập nhật tập dữ liệu. Người nghe cho phép người dùng Airflow tạo đăng ký cho các sự kiện nhất định xảy ra trong môi trường. Trong Airflow 2.8, trình nghe được thêm vào cho hai sự kiện bộ dữ liệu: on_dataset_created và on_dataset_changed, cho phép người dùng Airflow viết mã tùy chỉnh một cách hiệu quả để phản ứng với các hoạt động quản lý tập dữ liệu. Ví dụ: bạn có thể kích hoạt hệ thống bên ngoài hoặc gửi thông báo.

Việc sử dụng móc nghe cho bộ dữ liệu rất đơn giản. Hoàn thành các bước sau để tạo trình nghe cho on_dataset_changed:

1. Tạo trình nghe (dataset_listener.py):

   from airflow import Dataset
   from airflow.listeners import hookimpl
   
   @hookimpl
   def on_dataset_changed(dataset: Dataset):
       """Following custom code is executed when a dataset is changed."""
       print("Invoking external endpoint")
   
       """Validating a specific dataset"""
       if dataset.uri == "s3://bucket-prefix/object-key.ext":
           print ("Execute specific/different action for this dataset")

2. Tạo một plugin để đăng ký trình nghe trong môi trường Airflow của bạn (dataset_listener_plugin.py):

   from airflow.plugins_manager import AirflowPlugin
   from plugins import listener_code
   
   class DatasetListenerPlugin(AirflowPlugin):
       name = "dataset_listener_plugin"
       listeners = [dataset_listener]

Để biết thêm thông tin về cách cài đặt plugin trong Amazon MWAA, hãy tham khảo Cài đặt plugin tùy chỉnh.

Thiết lập môi trường Airflow 2.8.1 mới trong Amazon MWAA

Bạn có thể khởi tạo thiết lập trong tài khoản của bạn và Vùng ưa thích bằng cách sử dụng Bảng điều khiển quản lý AWS, API, hoặc Giao diện dòng lệnh AWS (AWS CLI). Nếu bạn đang sử dụng cơ sở hạ tầng dưới dạng mã (IaC), bạn có thể tự động hóa quá trình thiết lập bằng cách sử dụng Hình thành đám mây AWS, Các Bộ công cụ phát triển đám mây AWS (AWS CDK) hoặc tập lệnh Terraform.

Sau khi tạo thành công môi trường Airflow phiên bản 2.8.1 trong Amazon MWAA, một số gói nhất định sẽ được cài đặt tự động trên nút lập lịch trình và nút công việc. Để biết danh sách đầy đủ các gói đã cài đặt và phiên bản của chúng, hãy tham khảo Gói nhà cung cấp Apache Airflow được cài đặt trên môi trường Amazon MWAA. Bạn có thể cài đặt các gói bổ sung bằng tệp yêu cầu.

Nâng cấp từ các phiên bản cũ hơn của Airflow lên phiên bản 2.8.1

Bạn có thể tận dụng những khả năng mới nhất này bằng cách nâng cấp môi trường dựa trên Airflow phiên bản 2.x cũ lên phiên bản 2.8.1 bằng cách sử dụng các bản nâng cấp phiên bản tại chỗ. Để tìm hiểu thêm về nâng cấp phiên bản tại chỗ, hãy tham khảo Nâng cấp phiên bản Apache Airflow or Giới thiệu nâng cấp phiên bản tại chỗ với Amazon MWAA.

Kết luận

Trong bài đăng này, chúng tôi đã thảo luận về một số tính năng quan trọng được giới thiệu trong Airflow phiên bản 2.8, chẳng hạn như lưu trữ đối tượng, tab XCom mới được thêm vào chế độ xem lưới, ghi nhật ký ngữ cảnh tác vụ, móc nghe cho bộ dữ liệu và cách bạn có thể bắt đầu sử dụng chúng. Chúng tôi cũng cung cấp một số mã mẫu để hiển thị quá trình triển khai trong Amazon MWAA. Để biết danh sách đầy đủ các thay đổi, hãy tham khảo Ghi chú phát hành của Airflow.

Để biết thêm chi tiết và ví dụ về mã trên Amazon MWAA, hãy truy cập Hướng dẫn sử dụng Amazon MWAA và Amazon MWAA ví dụ GitHub repo.

Apache, Apache Airflow và Airflow là nhãn hiệu đã đăng ký hoặc nhãn hiệu của Quỹ phần mềm Apache ở Hoa Kỳ và / hoặc các quốc gia khác.

Về các tác giả

Mansi Bhutada là Kiến trúc sư Giải pháp ISV có trụ sở tại Hà Lan. Cô giúp khách hàng thiết kế và triển khai các giải pháp có kiến ​​trúc tốt trong AWS nhằm giải quyết các vấn đề kinh doanh của họ. Cô ấy đam mê phân tích dữ liệu và kết nối mạng. Ngoài công việc, cô thích thử nghiệm đồ ăn, chơi ném bóng và tham gia các trò chơi board vui nhộn.

Hernan García là Kiến trúc sư giải pháp cấp cao tại AWS có trụ sở tại Hà Lan. Anh làm việc trong ngành dịch vụ tài chính, hỗ trợ các doanh nghiệp áp dụng đám mây. Anh ấy đam mê công nghệ serverless, bảo mật và tuân thủ. Anh ấy thích dành thời gian với gia đình và bạn bè, đồng thời thử những món ăn mới từ nhiều nền ẩm thực khác nhau.

• Phân phối nội dung và PR được hỗ trợ bởi SEO. Được khuếch đại ngay hôm nay.
• PlatoData.Network Vertical Generative Ai. Trao quyền cho chính mình. Truy cập Tại đây.
• PlatoAiStream. Thông minh Web3. Kiến thức khuếch đại. Truy cập Tại đây.
• Trung tâmESG. Than đá, công nghệ sạch, Năng lượng, Môi trường Hệ mặt trời, Quản lý chất thải. Truy cập Tại đây.
• PlatoSức khỏe. Tình báo thử nghiệm lâm sàng và công nghệ sinh học. Truy cập Tại đây.
• nguồn: https://aws.amazon.com/blogs/big-data/introducing-amazon-mwaa-support-for-apache-airflow-version-2-8-1/