by Eric Johnson | on 22 NOV 2021 | in Amazon Managed Streaming for Apache Kafka (Amazon MSK), Serverless | Permalink |  Share

Bài viết này được viết bởi Adam Wagner, Kiến trúc sư Giải pháp Chủ chốt về Serverless.

Năm ngoái, AWS đã công bố hỗ trợ cho Amazon Managed Streaming for Apache Kafka (MSK) và các cụm Apache Kafka tự quản lý như event sources cho các hàm AWS Lambda. Hôm nay, AWS thêm một chỉ số OffsetLag mới vào các hàm Lambda với nguồn sự kiện MSK hoặc Apache Kafka tự quản lý.

Offset trong Apache Kafka là một số nguyên đánh dấu vị trí hiện tại của một người tiêu thụ. OffsetLag là sự khác biệt về offset giữa bản ghi cuối cùng được ghi vào chủ đề Kafka và bản ghi cuối cùng được xử lý bởi Lambda. Kafka biểu thị điều này dưới dạng số lượng bản ghi, không phải là một đơn vị thời gian. Chỉ số này cung cấp khả năng nhìn thấy xem hàm Lambda của bạn có theo kịp các bản ghi được thêm vào chủ đề mà nó đang xử lý hay không.

Blog này hướng dẫn sử dụng chỉ số OffsetLag cùng với các chỉ số Lambda và MSK khác để hiểu ứng dụng streaming của bạn và tối ưu hóa hàm Lambda của bạn.

Tổng quan

Trong ứng dụng ví dụ này, một nhà sản xuất viết các thông điệp vào một chủ đề trên cụm MSK là nguồn sự kiện cho một hàm Lambda. Mỗi thông điệp chứa một số và hàm Lambda tìm các yếu tố của số đó. Nó đầu ra số đầu vào và kết quả vào một bảng Amazon DynamoDB.

Tìm tất cả các yếu tố của một số là nhanh chóng nếu số đó nhỏ nhưng mất nhiều thời gian hơn cho các số lớn hơn. Sự khác biệt này có nghĩa là kích thước của số được ghi vào chủ đề MSK ảnh hưởng đến thời gian thực thi của hàm Lambda.

Kiến trúc ứng dụng ví dụ

1. Một khách hàng Kafka viết các thông điệp vào một chủ đề trong cụm MSK.
2. Nguồn sự kiện Lambda đánh giá chủ đề MSK thay mặt bạn để tìm các thông điệp mới và kích hoạt hàm Lambda của bạn với các lô thông điệp.
3. Hàm Lambda phân tách số trong mỗi thông điệp và sau đó ghi kết quả vào DynamoDB.

Trong ứng dụng này, một số yếu tố có thể góp phần vào sự chậm trễ về offset. Đầu tiên là số lượng và kích thước của các thông điệp. Nếu có nhiều thông điệp đến, Lambda có thể mất nhiều thời gian hơn để xử lý chúng. Các yếu tố khác là số lượng phân vùng trong chủ đề và số lượng hàm Lambda đồng thời xử lý thông điệp. Một giải thích đầy đủ về cách đồng thời của Lambda mở rộng với nguồn sự kiện MSK có trong tài liệu.

Nếu thời gian trung bình của hàm Lambda của bạn tăng, điều này cũng có tendance làm tăng offset lag. Độ trễ này có thể là do độ trễ trong một dịch vụ ở dưới, hoặc do sự phức tạp của các thông điệp đầu vào. Cuối cùng, nếu hàm Lambda của bạn gặp lỗi, nguồn sự kiện MSK sẽ thử lại bộ ghi chép giống nhau cho đến khi chúng thành công. Chức năng thử lại này cũng làm tăng offset lag.

Đo lường OffsetLag

Để hiểu cách chỉ số mới OffsetLag hoạt động, trước tiên bạn cần một chủ đề MSK hoạt động như một nguồn sự kiện cho một hàm Lambda. Theo dõi bài blog này để thiết lập một trường hợp MSK.

Để tìm chỉ số OffsetLag, truy cập vào bảng điều khiển CloudWatch, chọn Tất cả các chỉ số từ menu bên trái. Tiếp theo chọn Lambda, sau đó chọn Theo tên hàm để xem danh sách các chỉ số theo hàm Lambda. Cuộn hoặc sử dụng thanh tìm kiếm để tìm các chỉ số cho hàm này và chọn OffsetLag.

ví dụ về chỉ số offetLag

Để dễ dàng theo dõi nhiều chỉ số cùng một lúc, hãy tạo một bảng điều khiển CloudWatch bắt đầu với chỉ số OffsetLag. Chọn Actions -> Add to Dashboard. Chọn nút Create new, đặt tên cho bảng điều khiển. Chọn Tạo, giữ nguyên các tùy chọn còn lại ở mặc định.

Thêm OffsetLag vào bảng điều khiển

Sau khi chọn Add to dashboard, bảng điều khiển mới sẽ xuất hiện. Chọn nút Add widget để thêm chỉ số thời gian thực thi Lambda từ cùng một hàm. Thêm một tiện ích khác kết hợp cả lỗi Lambda và số lần gọi hàm cho hàm đó. Cuối cùng, thêm một tiện ích cho chỉ số BytesInPerSec cho chủ đề MSK. Tìm chỉ số này trong AWS/Kafka -> Broker ID, Cluster Name, Topic. Cuối cùng, nhấn Save dashboard.

Sau vài phút, bạn thấy một luồng liên tục các lời gọi, như bạn mong đợi khi tiêu thụ từ một chủ đề bận rộn.

Dữ liệu đang được gửi đến bảng điều khiển.

Trong ví dụ này, đây là một bảng điều khiển CloudWatch hiển thị các chỉ số OffsetLag, Duration, Errors và Invocations của Lambda, cùng với BytesInPerSec cho chủ đề MSK.

Trong ví dụ này, chỉ số OffsetLag trung bình khoảng tám, cho thấy rằng hàm Lambda đang chậm khoảng tám bản ghi so với bản ghi mới nhất trong chủ đề. Mặc dù điều này là chấp nhận được, nhưng vẫn có không gian để cải thiện.

Điều đầu tiên cần kiểm tra là lỗi của hàm Lambda, có thể làm tăng offset lag. Các chỉ số cho thấy không có lỗi nào, vì vậy bước tiếp theo là đánh giá và tối ưu hóa mã.

Hàm xử lý Lambda lặp qua các bản ghi và gọi hàm process_msg cho mỗi bản ghi:

def lambda_handler(event, context):     for batch in event[‘records’].keys():         for record in event[‘records’][batch]:             try:                 process_msg(record)             except:                 print(“error processing record:”, record)     return()

Hàm process_msg xử lý giải mã base64, gọi một hàm factor để phân tích số và ghi bản ghi vào bảng DynamoDB:

def process_msg(record):     #messages are base64 encoded, so we decode it here     msg_value = base64.b64decode(record[‘value’]).decode()     msg_dict = json.loads(msg_value)     #using the number as the hash key in the dynamodb table     msg_id = f”{msg_dict[‘number’]}”     if msg_dict[‘number’] <= MAX_NUMBER:         factors = factor_number(msg_dict[‘number’])         print(f”number: {msg_dict[‘number’]} has factors: {factors}”)         item = {‘msg_id’: msg_id, ‘msg’:msg_value, ‘factors’:factors}         resp = ddb_table.put_item(Item=item)     else:         print(f”ERROR: {msg_dict[‘number’]} is >= limit of {MAX_NUMBER}”)

Các tính toán nặng nề diễn ra trong hàm factor:

def factor(number):     factors = [1,number]     for x in range(2, (int(1 + number / 2))):         if (number % x) == 0:             factors.append(x)     return factors

Mã lặp qua tất cả các số cho đến khi số được phân tích chia cho hai. Mã được tối ưu hóa bằng cách chỉ lặp qua đến căn bậc hai của số.

def factor(number):     factors = [1,number]     for x in range(2, 1 + int(number**0.5)):         if (number % x) == 0:             factors.append(x)             factors.append(number // x)     return factors

Có thêm các tối ưu hóa và thư viện để phân tích số, nhưng điều này cung cấp một cải thiện hiệu suất đáng kể trong ví dụ này.

Dữ liệu sau khi tối ưu hóa

Sau khi triển khai mã, làm mới các chỉ số sau một thời gian để xem các cải tiến:

Thời gian trung bình của Lambda đã giảm xuống mili giây đơn chữ số và OffsetLag hiện đang trung bình là hai.

Nếu bạn thấy một sự thay đổi đáng kể trong chỉ số OffsetLag, có một số điều cần điều tra. Phía đầu vào của hệ thống, số lượng tin nhắn trên giây tăng lên, hoặc một sự tăng đáng kể trong kích thước của tin nhắn là một số lựa chọn.

Kết luận

Bài viết này hướng dẫn cách triển khai chỉ số OffsetLag để hiểu về độ trễ giữa các tin nhắn mới nhất trong chủ đề MSK và các bản ghi mà một hàm Lambda đang xử lý. Nó cũng đánh giá các chỉ số khác giúp hiểu nguyên nhân cơ bản của sự tăng offset lag. Để biết thêm thông tin về chủ đề này, vui lòng tham khảo tài liệu và các chỉ số MSK Lambda khác.

Để biết thêm nguồn tài nguyên học tập về serverless, hãy truy Serverless Land.