Apache Kafka - Hướng dẫn nhanh

Trong Dữ liệu lớn, một khối lượng dữ liệu khổng lồ được sử dụng. Về dữ liệu, chúng tôi có hai thách thức chính, thách thức đầu tiên là làm thế nào để thu thập khối lượng lớn dữ liệu và thách thức thứ hai là phân tích dữ liệu thu thập được. Để vượt qua những thách thức đó, bạn phải cần một hệ thống nhắn tin.

Kafka được thiết kế cho các hệ thống thông lượng cao phân tán. Kafka có xu hướng hoạt động rất hiệu quả khi thay thế cho một nhà môi giới tin nhắn truyền thống hơn. So với các hệ thống nhắn tin khác, Kafka có thông lượng tốt hơn, phân vùng tích hợp, sao chép và khả năng chịu lỗi vốn có, điều này làm cho nó phù hợp với các ứng dụng xử lý tin nhắn quy mô lớn.

Hệ thống nhắn tin là gì?

Hệ thống Nhắn tin chịu trách nhiệm chuyển dữ liệu từ ứng dụng này sang ứng dụng khác, vì vậy các ứng dụng có thể tập trung vào dữ liệu, nhưng không phải lo lắng về cách chia sẻ dữ liệu đó. Nhắn tin phân tán dựa trên khái niệm xếp hàng tin nhắn đáng tin cậy. Tin nhắn được xếp hàng đợi không đồng bộ giữa các ứng dụng khách và hệ thống nhắn tin. Có hai kiểu nhắn tin - một kiểu là trỏ tới điểm và kiểu kia là hệ thống nhắn tin đăng ký xuất bản (pub-sub). Hầu hết các kiểu nhắn tin tuân theopub-sub.

Hệ thống nhắn tin điểm đến điểm

Trong hệ thống điểm-điểm, các thông báo được lưu trong một hàng đợi. Một hoặc nhiều người tiêu dùng có thể sử dụng các tin nhắn trong hàng đợi, nhưng một tin nhắn cụ thể chỉ có thể được sử dụng bởi tối đa một người tiêu dùng. Khi người tiêu dùng đọc một tin nhắn trong hàng đợi, tin nhắn đó sẽ biến mất khỏi hàng đợi đó. Ví dụ điển hình của hệ thống này là Hệ thống xử lý đơn đặt hàng, trong đó mỗi đơn đặt hàng sẽ được xử lý bởi một Bộ xử lý đơn hàng, nhưng nhiều Bộ xử lý đơn hàng cũng có thể hoạt động cùng một lúc. Sơ đồ sau đây mô tả cấu trúc.

Xuất bản-Đăng ký Hệ thống nhắn tin

Trong hệ thống đăng ký xuất bản, các bài viết vẫn tồn tại trong một chủ đề. Không giống như hệ thống điểm-điểm, người tiêu dùng có thể đăng ký một hoặc nhiều chủ đề và xem tất cả các thông điệp trong chủ đề đó. Trong hệ thống Đăng ký-Đăng ký, nhà sản xuất tin nhắn được gọi là nhà xuất bản và người tiêu dùng tin nhắn được gọi là người đăng ký. Một ví dụ thực tế là Dish TV, nơi xuất bản các kênh khác nhau như thể thao, phim, âm nhạc, v.v. và bất kỳ ai cũng có thể đăng ký nhóm kênh của riêng mình và tải chúng bất cứ khi nào có kênh đã đăng ký của họ.

Kafka là gì?

Apache Kafka là một hệ thống nhắn tin đăng ký xuất bản phân tán và một hàng đợi mạnh mẽ có thể xử lý một lượng lớn dữ liệu và cho phép bạn chuyển các tin nhắn từ điểm cuối này đến điểm cuối khác. Kafka thích hợp cho cả việc sử dụng tin nhắn ngoại tuyến và trực tuyến. Các thông báo Kafka vẫn còn trên đĩa và được sao chép trong cụm để tránh mất dữ liệu. Kafka được xây dựng dựa trên dịch vụ đồng bộ hóa ZooKeeper. Nó tích hợp rất tốt với Apache Storm và Spark để phân tích dữ liệu phát trực tuyến theo thời gian thực.

Những lợi ích

Sau đây là một số lợi ích của Kafka:

Reliability - Kafka được phân phối, phân vùng, nhân rộng và khả năng chịu lỗi.
Scalability - Hệ thống nhắn tin Kafka mở rộng quy mô dễ dàng mà không mất thời gian ..
Durability- Kafka sử dụng Nhật ký cam kết phân tán có nghĩa là các thông báo sẽ tồn tại trên đĩa nhanh nhất có thể, do đó nó bền ..
Performance- Kafka có thông lượng cao cho cả việc xuất bản và đăng ký tin nhắn. Nó duy trì hiệu suất ổn định ngay cả khi nhiều TB tin nhắn được lưu trữ.

Kafka rất nhanh và đảm bảo không mất thời gian chết và không mất dữ liệu.

Trường hợp sử dụng

Kafka có thể được sử dụng trong nhiều Trường hợp sử dụng. Một số trong số chúng được liệt kê dưới đây -

Metrics- Kafka thường được sử dụng cho dữ liệu giám sát hoạt động. Điều này liên quan đến việc tổng hợp thống kê từ các ứng dụng phân tán để tạo ra nguồn cấp dữ liệu hoạt động tập trung.
Log Aggregation Solution - Kafka có thể được sử dụng trong một tổ chức để thu thập nhật ký từ nhiều dịch vụ và cung cấp chúng ở định dạng tiêu chuẩn cho nhiều kẻ lừa đảo.
Stream Processing- Các khuôn khổ phổ biến như Storm và Spark Streaming đọc dữ liệu từ một chủ đề, xử lý nó và ghi dữ liệu đã xử lý vào một chủ đề mới mà nó có sẵn cho người dùng và ứng dụng. Độ bền mạnh mẽ của Kafka cũng rất hữu ích trong bối cảnh xử lý dòng.

Cần cho Kafka

Kafka là một nền tảng thống nhất để xử lý tất cả các nguồn cấp dữ liệu thời gian thực. Kafka hỗ trợ gửi tin nhắn có độ trễ thấp và đảm bảo khả năng chịu lỗi trong trường hợp máy bị lỗi. Nó có khả năng xử lý một số lượng lớn người tiêu dùng đa dạng. Kafka rất nhanh, thực hiện 2 triệu lần ghi / giây. Kafka lưu giữ tất cả dữ liệu vào đĩa, về cơ bản có nghĩa là tất cả các lần ghi đều được chuyển vào bộ đệm trang của hệ điều hành (RAM). Điều này làm cho việc chuyển dữ liệu từ bộ đệm trang sang ổ cắm mạng rất hiệu quả.

Trước khi đi sâu vào Kafka, bạn phải biết các thuật ngữ chính như chủ đề, nhà môi giới, nhà sản xuất và người tiêu dùng. Sơ đồ sau minh họa các thuật ngữ chính và bảng mô tả chi tiết các thành phần của sơ đồ.

Trong sơ đồ trên, một chủ đề được cấu hình thành ba phân vùng. Phân vùng 1 có hai hệ số bù 0 và 1. Phân vùng 2 có bốn hệ số bù 0, 1, 2 và 3. Phân vùng 3 có một hệ số bù 0. Id của bản sao giống với id của máy chủ lưu trữ nó.

Giả sử, nếu hệ số nhân bản của chủ đề được đặt thành 3, thì Kafka sẽ tạo ra 3 bản sao giống hệt nhau của mỗi phân vùng và đặt chúng vào cụm để sẵn sàng cho tất cả các hoạt động của nó. Để cân bằng tải trong cụm, mỗi nhà môi giới lưu trữ một hoặc nhiều phân vùng đó. Nhiều nhà sản xuất và người tiêu dùng có thể xuất bản và truy xuất các thông điệp cùng một lúc.

S.Không	Các thành phần và mô tả
1	Topics Một luồng thông báo thuộc một danh mục cụ thể được gọi là chủ đề. Dữ liệu được lưu trữ trong các chủ đề. Các chủ đề được chia thành các phân vùng. Đối với mỗi chủ đề, Kafka giữ một phân vùng nhỏ. Mỗi phân vùng như vậy chứa các thông điệp trong một trình tự có thứ tự bất biến. Một phân vùng được triển khai dưới dạng một tập hợp các tệp phân đoạn có kích thước bằng nhau.
2	Partition Chủ đề có thể có nhiều phân vùng, vì vậy nó có thể xử lý một lượng dữ liệu tùy ý.
3	Partition offset Mỗi thông báo được phân vùng có một id chuỗi duy nhất được gọi là offset .
4	Replicas of partition Bản sao không là gì ngoài bản sao lưu của một phân vùng. Bản sao không bao giờ đọc hoặc ghi dữ liệu. Chúng được sử dụng để ngăn chặn mất dữ liệu.
5	Brokers Các nhà môi giới là hệ thống đơn giản chịu trách nhiệm duy trì dữ liệu công khai. Mỗi nhà môi giới có thể có không hoặc nhiều phân vùng cho mỗi chủ đề. Giả sử, nếu có N phân vùng trong một chủ đề và N số nhà môi giới, mỗi nhà môi giới sẽ có một phân vùng. Giả sử nếu có N phân vùng trong một chủ đề và nhiều hơn N môi giới (n + m), N môi giới đầu tiên sẽ có một phân vùng và M môi giới tiếp theo sẽ không có bất kỳ phân vùng nào cho chủ đề cụ thể đó. Giả sử nếu có N phân vùng trong một chủ đề và ít hơn N môi giới (nm), mỗi nhà môi giới sẽ có một hoặc nhiều phân vùng chia sẻ giữa chúng. Kịch bản này không được khuyến nghị do phân phối tải không đồng đều giữa các nhà môi giới.
6	Kafka Cluster Kafka có nhiều hơn một nhà môi giới được gọi là Kafka cluster. Một cụm Kafka có thể được mở rộng mà không cần thời gian chết. Các cụm này được sử dụng để quản lý sự tồn tại và nhân rộng của dữ liệu tin nhắn.
7	Producers Nhà sản xuất là người xuất bản thông báo cho một hoặc nhiều chủ đề Kafka. Các nhà sản xuất gửi dữ liệu đến các nhà môi giới Kafka. Mỗi khi nhà sản xuất pub-lishes một thông điệp cho một nhà môi giới, nhà môi giới chỉ cần thêm thông báo vào tệp phân đoạn cuối cùng. Trên thực tế, thông báo sẽ được thêm vào một phân vùng. Nhà sản xuất cũng có thể gửi tin nhắn đến một phân vùng mà họ lựa chọn.
số 8	Consumers Người tiêu dùng đọc dữ liệu từ các nhà môi giới. Người tiêu dùng đăng ký một hoặc nhiều chủ đề và xem các thông điệp đã xuất bản bằng cách lấy dữ liệu từ các nhà môi giới.
9	Leader Leader là nút chịu trách nhiệm cho tất cả các lần đọc và ghi cho phân vùng đã cho. Mỗi phân vùng có một máy chủ hoạt động như một nhà lãnh đạo.
10	Follower Nút tuân theo hướng dẫn của người lãnh đạo được gọi là người theo dõi. Nếu người lãnh đạo không thành công, một trong những người đi theo sẽ tự động trở thành người lãnh đạo mới. Một người theo dõi hoạt động như một người tiêu dùng bình thường, lấy tin nhắn và cập nhật kho dữ liệu của chính họ.

Hãy xem hình minh họa sau đây. Nó cho thấy sơ đồ cụm của Kafka.

Bảng sau mô tả từng thành phần được hiển thị trong sơ đồ trên.

S.Không	Các thành phần và mô tả
1	Broker Cụm Kafka thường bao gồm nhiều nhà môi giới để duy trì cân bằng tải. Các nhà môi giới Kafka không có trạng thái, vì vậy họ sử dụng ZooKeeper để duy trì trạng thái cụm của họ. Một phiên bản của nhà môi giới Kafka có thể xử lý hàng trăm nghìn lần đọc và ghi mỗi giây và mỗi bro-ker có thể xử lý hàng TB thư mà không ảnh hưởng đến hiệu suất. Cuộc bầu cử lãnh đạo nhà môi giới Kafka có thể được thực hiện bởi ZooKeeper.
2	ZooKeeper ZooKeeper được sử dụng để quản lý và điều phối nhà môi giới Kafka. Dịch vụ ZooKeeper chủ yếu được sử dụng để thông báo cho nhà sản xuất và người tiêu dùng về sự hiện diện của bất kỳ nhà môi giới mới nào trong hệ thống Kafka hoặc sự cố của nhà môi giới trong hệ thống Kafka. Theo thông báo mà Zookeeper nhận được liên quan đến sự hiện diện hay thất bại của nhà môi giới thì người chuyên nghiệp và người tiêu dùng sẽ đưa ra quyết định và bắt đầu phối hợp nhiệm vụ của họ với một số nhà môi giới khác.
3	Producers Nhà sản xuất đẩy dữ liệu đến các nhà môi giới. Khi nhà môi giới mới được bắt đầu, tất cả các nhà sản xuất tìm kiếm nó và tự động gửi tin nhắn đến nhà môi giới mới đó. Nhà sản xuất Kafka không chờ đợi xác nhận từ nhà môi giới và gửi tin nhắn nhanh nhất có thể xử lý.
4	Consumers Vì các nhà môi giới Kafka là không quốc tịch, có nghĩa là người tiêu dùng phải duy trì số lượng tin nhắn đã được sử dụng bằng cách sử dụng bù phân vùng. Nếu người tiêu dùng thừa nhận một phần bù tin nhắn cụ thể, điều đó có nghĩa là người tiêu dùng đã sử dụng tất cả các tin nhắn trước đó. Người tiêu dùng đưa ra một yêu cầu kéo không đồng bộ đến người môi giới để có một bộ đệm byte sẵn sàng để tiêu thụ. Người tiêu dùng có thể tua lại hoặc bỏ qua đến bất kỳ điểm nào trong phân vùng chỉ bằng cách cung cấp giá trị bù đắp. Giá trị bù đắp của người tiêu dùng được ZooKeeper thông báo.

Hiện tại, chúng ta đã thảo luận về các khái niệm cốt lõi của Kafka. Bây giờ chúng ta hãy làm sáng tỏ quy trình làm việc của Kafka.

Kafka chỉ đơn giản là một tập hợp các chủ đề được chia thành một hoặc nhiều phân vùng. Phân vùng Kafka là một chuỗi thông báo được sắp xếp theo thứ tự tuyến tính, trong đó mỗi thông báo được xác định bằng chỉ mục của chúng (được gọi là độ lệch). Tất cả dữ liệu trong một cụm Kafka là sự kết hợp rời rạc của các phân vùng. Các tin nhắn đến được viết ở cuối một phân vùng và các tin nhắn được người tiêu dùng đọc tuần tự. Độ bền được cung cấp bằng cách sao chép thông điệp đến các nhà môi giới khác nhau.

Kafka cung cấp cả hệ thống nhắn tin dựa trên pub-sub và queue theo cách thức nhanh chóng, đáng tin cậy, bền bỉ, khả năng chịu lỗi và không có thời gian chết. Trong cả hai trường hợp, nhà sản xuất chỉ cần gửi thông điệp đến một chủ đề và người tiêu dùng có thể chọn bất kỳ một loại hệ thống nhắn tin nào tùy thuộc vào nhu cầu của họ. Chúng ta hãy làm theo các bước trong phần tiếp theo để hiểu cách người tiêu dùng có thể chọn hệ thống nhắn tin mà họ chọn.

Luồng công việc của Nhắn tin Pub-Sub

Sau đây là quy trình làm việc khôn ngoan từng bước của Nhắn tin Pub-Sub -

Các nhà sản xuất gửi tin nhắn đến một chủ đề theo định kỳ.
Nhà môi giới Kafka lưu trữ tất cả các thông báo trong các phân vùng được định cấu hình cho chủ đề cụ thể đó. Nó đảm bảo các thông điệp được chia sẻ đồng đều giữa các phân vùng. Nếu nhà sản xuất gửi hai tin nhắn và có hai phân vùng, Kafka sẽ lưu trữ một tin nhắn trong phân vùng thứ nhất và tin nhắn thứ hai trong phân vùng thứ hai.
Người tiêu dùng đăng ký một chủ đề cụ thể.
Khi người tiêu dùng đăng ký một chủ đề, Kafka sẽ cung cấp phần bù hiện tại của chủ đề đó cho người tiêu dùng và cũng tiết kiệm phần bù vào nhóm Zookeeper.
Người tiêu dùng sẽ yêu cầu Kafka trong khoảng thời gian đều đặn (như 100 Ms) cho các tin nhắn mới.
Khi Kafka nhận được tin nhắn từ nhà sản xuất, nó sẽ chuyển những tin nhắn này đến người tiêu dùng.
Người tiêu dùng sẽ nhận được tin nhắn và xử lý nó.
Sau khi tin nhắn được xử lý, người tiêu dùng sẽ gửi xác nhận đến nhà môi giới Kafka.
Khi Kafka nhận được một xác nhận, nó sẽ thay đổi giá trị bù đắp thành giá trị mới và cập nhật nó trong Zookeeper. Vì hiệu số được duy trì trong Zookeeper, người tiêu dùng có thể đọc chính xác thông báo tiếp theo ngay cả khi máy chủ ngừng hoạt động.
Luồng trên sẽ lặp lại cho đến khi người tiêu dùng ngừng yêu cầu.
Người tiêu dùng có tùy chọn tua lại / bỏ qua đến độ lệch mong muốn của một chủ đề bất kỳ lúc nào và đọc tất cả các tin nhắn tiếp theo.

Quy trình làm việc của Nhắn tin Hàng đợi / Nhóm Người tiêu dùng

Trong hệ thống nhắn tin hàng đợi thay vì một người tiêu dùng, một nhóm người tiêu dùng có cùng ID nhóm sẽ đăng ký một chủ đề. Nói một cách dễ hiểu, người tiêu dùng đăng ký một chủ đề có cùng ID nhóm được coi là một nhóm duy nhất và thông điệp được chia sẻ giữa họ. Hãy để chúng tôi kiểm tra quy trình làm việc thực tế của hệ thống này.

Các nhà sản xuất gửi thông điệp đến một chủ đề trong một khoảng thời gian đều đặn.
Kafka lưu trữ tất cả các thông báo trong các phân vùng được cấu hình cho chủ đề cụ thể đó tương tự như kịch bản trước đó.
Một người tiêu dùng đăng ký một chủ đề cụ thể, giả sử Chủ đề-01 với ID nhóm là Nhóm-1 .
Kafka tương tác với người tiêu dùng theo cách giống như Nhắn tin Pub-Sub cho đến khi người tiêu dùng mới đăng ký cùng một chủ đề, Chủ đề-01 với cùng ID Nhóm như Nhóm-1 .
Khi người tiêu dùng mới đến, Kafka sẽ chuyển hoạt động của mình sang chế độ chia sẻ và chia sẻ dữ liệu giữa hai người tiêu dùng. Việc chia sẻ này sẽ tiếp tục cho đến khi số người phản bác đạt đến số lượng phân vùng được định cấu hình cho chủ đề cụ thể đó.
Khi số lượng người tiêu dùng vượt quá số lượng phân vùng, người tiêu dùng mới sẽ không nhận được bất kỳ thông báo nào nữa cho đến khi bất kỳ người tiêu dùng hiện tại nào hủy đăng ký. Kịch bản này nảy sinh bởi vì mỗi người tiêu dùng ở Kafka sẽ được chỉ định tối thiểu một phân vùng và một khi tất cả các phân vùng được chỉ định cho những người tiêu dùng hiện tại, những người tiêu dùng mới sẽ phải đợi.
Tính năng này còn được gọi là Nhóm người tiêu dùng . Theo cách tương tự, Kafka sẽ cung cấp những gì tốt nhất của cả hai hệ thống một cách rất đơn giản và hiệu quả.

Vai trò của ZooKeeper

Một phụ thuộc quan trọng của Apache Kafka là Apache Zookeeper, là một dịch vụ đồng bộ hóa và cấu hình phân tán. Zookeeper đóng vai trò là giao diện phối hợp giữa các nhà môi giới Kafka và người tiêu dùng. Các máy chủ Kafka chia sẻ thông tin qua một cụm Zookeeper. Kafka lưu trữ siêu dữ liệu cơ bản trong Zookeeper, chẳng hạn như thông tin về chủ đề, nhà môi giới, phần bù của người tiêu dùng (người đọc hàng đợi), v.v.

Vì tất cả thông tin quan trọng được lưu trữ trong Zookeeper và nó thường sao chép dữ liệu này trên toàn bộ nhóm của nó, lỗi của Kafka broker / Zookeeper không ảnh hưởng đến trạng thái của cụm Kafka. Kafka sẽ khôi phục trạng thái khi Zookeeper khởi động lại. Điều này giúp cho Kafka không có thời gian chết. Bầu cử lãnh đạo giữa nhà môi giới Kafka cũng được thực hiện bằng cách sử dụng Zookeeper trong trường hợp lãnh đạo thất bại.

Để tìm hiểu thêm về Zookeeper, vui lòng tham khảo Zookeeper

Hãy để chúng tôi tiếp tục tìm hiểu thêm về cách cài đặt Java, ZooKeeper và Kafka trên máy của bạn trong chương tiếp theo.

Sau đây là các bước để cài đặt Java trên máy của bạn.

Bước 1 - Xác minh cài đặt Java

Hy vọng rằng bạn đã cài đặt java trên máy tính của mình ngay bây giờ, vì vậy bạn chỉ cần xác minh nó bằng cách sử dụng lệnh sau.

$ java -version

Nếu java được cài đặt thành công trên máy của bạn, bạn có thể thấy phiên bản Java đã cài đặt.

Bước 1.1 - Tải xuống JDK

Nếu Java chưa được tải xuống, vui lòng tải xuống phiên bản mới nhất của JDK bằng cách truy cập liên kết sau và tải xuống phiên bản mới nhất.

http://www.oracle.com/technetwork/java/javase/downloads/index.html

Hiện tại, phiên bản mới nhất là JDK 8u 60 và tệp là “jdk-8u60-linux-x64.tar.gz”. Vui lòng tải xuống tệp trên máy của bạn.

Bước 1.2 - Giải nén tệp

Nói chung, các tệp đang được tải xuống được lưu trữ trong thư mục tải xuống, hãy xác minh nó và giải nén thiết lập tar bằng các lệnh sau.

$ cd /go/to/download/path
$ tar -zxf jdk-8u60-linux-x64.gz

Bước 1.3 - Di chuyển đến Opt Directory

Để cung cấp java cho tất cả người dùng, hãy di chuyển nội dung java đã giải nén vào thư mục usr / local / java /.

$ su
password: (type password of root user)
$ mkdir /opt/jdk $ mv jdk-1.8.0_60 /opt/jdk/

Bước 1.4 - Đặt đường dẫn

Để đặt đường dẫn và các biến JAVA_HOME, hãy thêm các lệnh sau vào tệp ~ / .bashrc.

export JAVA_HOME =/usr/jdk/jdk-1.8.0_60
export PATH=$PATH:$JAVA_HOME/bin

Bây giờ áp dụng tất cả các thay đổi vào hệ thống đang chạy hiện tại.

$ source ~/.bashrc

Bước 1.5 - Các lựa chọn thay thế Java

Sử dụng lệnh sau để thay đổi Java Alternatives.

update-alternatives --install /usr/bin/java java /opt/jdk/jdk1.8.0_60/bin/java 100

Step 1.6 - Bây giờ xác minh java bằng lệnh xác minh (java -version) được giải thích ở Bước 1.

Bước 2 - Cài đặt ZooKeeper Framework

Bước 2.1 - Tải xuống ZooKeeper

Để cài đặt ZooKeeper framework trên máy của bạn, hãy truy cập liên kết sau và tải xuống phiên bản ZooKeeper mới nhất.

http://zookeeper.apache.org/releases.html

Hiện tại, phiên bản mới nhất của ZooKeeper là 3.4.6 (ZooKeeper-3.4.6.tar.gz).

Bước 2.2 - Giải nén tệp tar

Giải nén tệp tar bằng lệnh sau

$ cd opt/
$ tar -zxf zookeeper-3.4.6.tar.gz $ cd zookeeper-3.4.6
$ mkdir data

Bước 2.3 - Tạo tệp cấu hình

Mở tệp cấu hình có tên conf / zoo.cfg bằng lệnh vi “conf / zoo.cfg” và tất cả các tham số sau để đặt làm điểm bắt đầu.

$ vi conf/zoo.cfg
tickTime=2000
dataDir=/path/to/zookeeper/data
clientPort=2181
initLimit=5
syncLimit=2

Khi tệp cấu hình đã được lưu thành công và quay trở lại thiết bị đầu cuối, bạn có thể khởi động máy chủ của người nuôi thú.

Bước 2.4 - Khởi động Máy chủ ZooKeeper

$ bin/zkServer.sh start

Sau khi thực hiện lệnh này, bạn sẽ nhận được phản hồi như hình dưới đây:

$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg $ Starting zookeeper ... STARTED

Bước 2.5 - Khởi động CLI

$ bin/zkCli.sh

Sau khi gõ lệnh trên, bạn sẽ được kết nối với máy chủ của Zookeeper và sẽ nhận được phản hồi bên dưới.

Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]

Bước 2.6 - Dừng máy chủ Zookeeper

Sau khi kết nối máy chủ và thực hiện tất cả các thao tác, bạn có thể dừng máy chủ sở thú bằng lệnh sau:

$ bin/zkServer.sh stop

Bây giờ bạn đã cài đặt thành công Java và ZooKeeper trên máy của mình. Hãy để chúng tôi xem các bước để cài đặt Apache Kafka.

Bước 3 - Cài đặt Apache Kafka

Hãy để chúng tôi tiếp tục các bước sau để cài đặt Kafka trên máy của bạn.

Bước 3.1 - Tải xuống Kafka

Để cài đặt Kafka trên máy của bạn, hãy nhấp vào liên kết dưới đây -

https://www.apache.org/dyn/closer.cgi?path=/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz

Bây giờ là phiên bản mới nhất, tức là - kafka_2.11_0.9.0.0.tgz sẽ được tải xuống máy của bạn.

Bước 3.2 - Giải nén tệp tar

Giải nén tệp tar bằng lệnh sau:

$ cd opt/ $ tar -zxf kafka_2.11.0.9.0.0 tar.gz
$ cd kafka_2.11.0.9.0.0

Bây giờ bạn đã tải xuống phiên bản Kafka mới nhất trên máy của mình.

Bước 3.3 - Khởi động máy chủ

Bạn có thể khởi động máy chủ bằng cách đưa ra lệnh sau:

$ bin/kafka-server-start.sh config/server.properties

Sau khi máy chủ khởi động, bạn sẽ thấy phản hồi dưới đây trên màn hình của mình -

$ bin/kafka-server-start.sh config/server.properties
[2016-01-02 15:37:30,410] INFO KafkaConfig values:
request.timeout.ms = 30000
log.roll.hours = 168
inter.broker.protocol.version = 0.9.0.X
log.preallocate = false
security.inter.broker.protocol = PLAINTEXT
…………………………………………….
…………………………………………….

Bước 4 - Dừng máy chủ

Sau khi thực hiện tất cả các thao tác, bạn có thể dừng máy chủ bằng lệnh sau:

$ bin/kafka-server-stop.sh config/server.properties

Bây giờ chúng ta đã thảo luận về cài đặt Kafka, chúng ta có thể tìm hiểu cách thực hiện các thao tác cơ bản trên Kafka trong chương tiếp theo.

Trước tiên, chúng ta hãy bắt đầu triển khai cấu hình nhà môi giới một nút đơn và sau đó chúng tôi sẽ chuyển thiết lập của mình sang cấu hình nhà môi giới nhiều nút duy nhất.

Hy vọng rằng bây giờ bạn đã cài đặt Java, ZooKeeper và Kafka trên máy của mình. Trước khi chuyển sang Thiết lập cụm Kafka, trước tiên bạn cần khởi động ZooKeeper của mình vì Kafka Cluster sử dụng ZooKeeper.

Khởi động ZooKeeper

Mở một thiết bị đầu cuối mới và nhập lệnh sau:

bin/zookeeper-server-start.sh config/zookeeper.properties

Để khởi động Kafka Broker, hãy gõ lệnh sau:

bin/kafka-server-start.sh config/server.properties

Sau khi khởi động Kafka Broker, gõ lệnh JPS trên Zookeeper thiết bị đầu cuối và bạn sẽ thấy các phản ứng sau -

821 QuorumPeerMain
928 Kafka
931 Jps

Bây giờ bạn có thể thấy hai daemon chạy trên thiết bị đầu cuối trong đó QuorumPeerMain là daemon ZooKeeper và một con khác là daemon Kafka.

Cấu hình một Node-Single Broker

Trong cấu hình này, bạn có một phiên bản ZooKeeper và id người môi giới. Sau đây là các bước để cấu hình nó:

Creating a Kafka Topic- Kafka cung cấp tiện ích dòng lệnh có tên kafka-topics.sh để tạo các chủ đề trên máy chủ. Mở thiết bị đầu cuối mới và nhập ví dụ dưới đây.

Syntax

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 
--partitions 1 --topic topic-name

Example

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1   
--partitions 1 --topic Hello-Kafka

Chúng tôi vừa tạo một chủ đề có tên Hello-Kafka với một phân vùng duy nhất và một nhân tố sao chép. Đầu ra được tạo ở trên sẽ tương tự như đầu ra sau:

Output- Đã tạo chủ đề Hello-Kafka

Khi chủ đề đã được tạo, bạn có thể nhận được thông báo trong cửa sổ đầu cuối của nhà môi giới Kafka và nhật ký cho chủ đề đã tạo được chỉ định trong “/ tmp / kafka-logs /“ trong tệp config / server.properties.

Danh sách các chủ đề

Để nhận danh sách các chủ đề trong máy chủ Kafka, bạn có thể sử dụng lệnh sau:

Syntax

bin/kafka-topics.sh --list --zookeeper localhost:2181

Output

Hello-Kafka

Vì chúng tôi đã tạo một chủ đề, nó sẽ chỉ liệt kê Hello-Kafka . Giả sử, nếu bạn tạo nhiều hơn một chủ đề, bạn sẽ nhận được tên chủ đề trong đầu ra.

Khởi động Producer để gửi tin nhắn

Syntax

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name

Từ cú pháp trên, hai tham số chính được yêu cầu cho máy khách dòng lệnh của nhà sản xuất -

Broker-list- Danh sách các nhà môi giới mà chúng tôi muốn gửi thông điệp đến. Trong trường hợp này, chúng tôi chỉ có một nhà môi giới. Tệp Config / server.properties chứa id cổng nhà môi giới, vì chúng tôi biết nhà môi giới của chúng tôi đang lắng nghe cổng 9092, vì vậy bạn có thể chỉ định trực tiếp.

Tên chủ đề - Đây là một ví dụ cho tên chủ đề.

Example

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka

Nhà sản xuất sẽ đợi đầu vào từ stdin và xuất bản lên cụm Kafka. Theo mặc định, mọi dòng mới được xuất bản dưới dạng một thông báo mới, sau đó các thuộc tính của nhà sản xuất mặc định được chỉ định trong tệp config / producer.properties . Bây giờ bạn có thể nhập một vài dòng thông báo trong terminal như hình dưới đây.

Output

$ bin/kafka-console-producer.sh --broker-list localhost:9092 
--topic Hello-Kafka[2016-01-16 13:50:45,931] 
WARN property topic is not valid (kafka.utils.Verifia-bleProperties)
Hello
My first message

My second message

Bắt đầu người tiêu dùng nhận tin nhắn

Tương tự như producer, các thuộc tính tiêu dùng mặc định được chỉ định trong tệp config / Consumer.proper-tie . Mở một thiết bị đầu cuối mới và gõ cú pháp dưới đây để sử dụng tin nhắn.

Syntax

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name 
--from-beginning

Example

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka 
--from-beginning

Output

Hello
My first message
My second message

Cuối cùng, bạn có thể nhập tin nhắn từ thiết bị đầu cuối của nhà sản xuất và thấy chúng xuất hiện trong thiết bị đầu cuối của người tiêu dùng. Hiện tại, bạn đã hiểu rất rõ về cụm nút đơn với một nhà môi giới duy nhất. Bây giờ chúng ta hãy chuyển sang cấu hình nhiều nhà môi giới.

Cấu hình một nút-nhiều nhà môi giới

Trước khi chuyển sang thiết lập cụm nhiều nhà môi giới, trước tiên hãy khởi động máy chủ ZooKeeper của bạn.

Create Multiple Kafka Brokers- Chúng tôi có một phiên bản người môi giới Kafka đã có trong con-fig / server.properties. Bây giờ chúng ta cần nhiều phiên bản môi giới, vì vậy hãy sao chép tệp server.prop-erties hiện có thành hai tệp cấu hình mới và đổi tên nó thành server-one.properties và server-two.prop-erties. Sau đó, chỉnh sửa cả hai tệp mới và chỉ định các thay đổi sau:

config / server-one.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
# The port the socket server listens on
port=9093
# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-1

config / server-two.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=2
# The port the socket server listens on
port=9094
# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-2

Start Multiple Brokers- Sau khi tất cả các thay đổi đã được thực hiện trên ba máy chủ, hãy mở ba thiết bị đầu cuối mới để bắt đầu từng nhà môi giới một.

Broker1
bin/kafka-server-start.sh config/server.properties
Broker2
bin/kafka-server-start.sh config/server-one.properties
Broker3
bin/kafka-server-start.sh config/server-two.properties

Bây giờ chúng tôi có ba nhà môi giới khác nhau đang chạy trên máy. Hãy thử một mình để kiểm tra tất cả các daemon bằng cách gõjps trên thiết bị đầu cuối ZooKeeper, sau đó bạn sẽ thấy phản hồi.

Tạo chủ đề

Hãy để chúng tôi gán giá trị hệ số nhân rộng là ba cho chủ đề này vì chúng tôi có ba nhà môi giới khác nhau đang hoạt động. Nếu bạn có hai nhà môi giới, thì giá trị bản sao được chỉ định sẽ là hai.

Syntax

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 
-partitions 1 --topic topic-name

Example

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 
-partitions 1 --topic Multibrokerapplication

Output

created topic “Multibrokerapplication”

Lệnh Mô tả được sử dụng để kiểm tra nhà môi giới nào đang lắng nghe chủ đề được tạo hiện tại như hình dưới đây:

bin/kafka-topics.sh --describe --zookeeper localhost:2181 
--topic Multibrokerappli-cation

Output

bin/kafka-topics.sh --describe --zookeeper localhost:2181 
--topic Multibrokerappli-cation

Topic:Multibrokerapplication    PartitionCount:1 
ReplicationFactor:3 Configs:
   
Topic:Multibrokerapplication Partition:0 Leader:0 
Replicas:0,2,1 Isr:0,2,1

Từ kết quả trên, chúng ta có thể kết luận rằng dòng đầu tiên cung cấp một bản tóm tắt về tất cả các phân vùng, hiển thị tên chủ đề, số lượng phân vùng và hệ số sao chép mà chúng ta đã chọn. Trong dòng thứ hai, mỗi nút sẽ dẫn đầu cho một phần được chọn ngẫu nhiên của các phân vùng.

Trong trường hợp của chúng tôi, chúng tôi thấy rằng nhà môi giới đầu tiên của chúng tôi (với broker.id 0) là người dẫn đầu. Sau đó Replicas: 0,2,1 có nghĩa là tất cả các nhà môi giới sao chép chủ đề cuối cùng Isr là tập hợp các bản sao đồng bộ . Chà, đây là tập hợp con của các bản sao hiện còn sống và bị bắt bởi kẻ đứng đầu.

Khởi động Producer để gửi tin nhắn

Quy trình này vẫn giống như trong thiết lập nhà môi giới đơn lẻ.

Example

bin/kafka-console-producer.sh --broker-list localhost:9092 
--topic Multibrokerapplication

Output

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication
[2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties)
This is single node-multi broker demo
This is the second message

Bắt đầu người tiêu dùng nhận tin nhắn

Quy trình này vẫn giống như được hiển thị trong thiết lập nhà môi giới duy nhất.

Example

bin/kafka-console-consumer.sh --zookeeper localhost:2181 
—topic Multibrokerapplica-tion --from-beginning

Output

bin/kafka-console-consumer.sh --zookeeper localhost:2181 
—topic Multibrokerapplica-tion —from-beginning
This is single node-multi broker demo
This is the second message

Hoạt động chủ đề cơ bản

Trong chương này, chúng ta sẽ thảo luận về các hoạt động chủ đề cơ bản khác nhau.

Sửa đổi chủ đề

Như bạn đã hiểu cách tạo chủ đề trong Kafka Cluster. Bây giờ chúng ta hãy sửa đổi chủ đề đã tạo bằng lệnh sau

Syntax

bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name 
--parti-tions count

Example

We have already created a topic “Hello-Kafka” with single partition count and one replica factor. 
Now using “alter” command we have changed the partition count.
bin/kafka-topics.sh --zookeeper localhost:2181 
--alter --topic Hello-kafka --parti-tions 2

Output

WARNING: If partitions are increased for a topic that has a key, 
the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

Xóa chủ đề

Để xóa một chủ đề, bạn có thể sử dụng cú pháp sau.

Syntax

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name

Example

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka

Output

> Topic Hello-kafka marked for deletion

Note −Điều này sẽ không có tác động nếu delete.topic.enable không được đặt thành true

Hãy để chúng tôi tạo một ứng dụng để xuất bản và sử dụng thư bằng ứng dụng Java. Ứng dụng khách nhà sản xuất Kafka bao gồm các API sau.

API của nhà sản xuất Kafka

Hãy để chúng tôi hiểu bộ API nhà sản xuất Kafka quan trọng nhất trong phần này. Phần trung tâm của API KafkaProduction là lớp KafkaProductioner . Lớp KafkaProductioner cung cấp một tùy chọn để kết nối một nhà môi giới Kafka trong phương thức khởi tạo của nó với các phương thức sau.

Lớp KafkaProductioner cung cấp phương thức gửi để gửi các thông điệp không đồng bộ đến một chủ đề. Chữ ký của send () như sau

producer.send(new ProducerRecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

ProducerRecord - Nhà sản xuất quản lý một bộ đệm các bản ghi đang chờ được gửi đi.
Callback - Lệnh gọi lại do người dùng cung cấp để thực hiện khi bản ghi đã được máy chủ xác nhận (null cho biết không có lệnh gọi lại).

Lớp KafkaProductioner cung cấp một phương thức tuôn ra để đảm bảo rằng tất cả các thư đã gửi trước đó đã thực sự được hoàn thành. Cú pháp của phương thức flush như sau:

public void flush()

Lớp KafkaProductioner cung cấp phương thức partitionFor, giúp lấy siêu dữ liệu phân vùng cho một chủ đề nhất định. Điều này có thể được sử dụng để phân vùng tùy chỉnh. Chữ ký của phương pháp này như sau:

public Map metrics()

Nó trả về bản đồ các chỉ số nội bộ do nhà sản xuất duy trì.

public void close () - Lớp KafkaProductioner cung cấp các khối phương thức đóng cho đến khi hoàn thành tất cả các yêu cầu đã gửi trước đó.

API nhà sản xuất

Phần trung tâm của API Producer là lớp Producer . Lớp Producer cung cấp một tùy chọn để kết nối Kafka broker trong constructor của nó bằng các phương thức sau.

Lớp nhà sản xuất

Lớp nhà sản xuất cung cấp phương thức gửi tới send tin nhắn cho một hoặc nhiều chủ đề bằng cách sử dụng các chữ ký sau.

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);

Có hai loại nhà sản xuất - Sync và Async.

Cấu hình API tương tự cũng áp dụng cho nhà sản xuất Đồng bộ hóa . Sự khác biệt giữa chúng là một nhà sản xuất đồng bộ gửi tin nhắn trực tiếp, nhưng gửi tin nhắn trong nền. Trình sản xuất không đồng bộ được ưu tiên khi bạn muốn thông lượng cao hơn. Trong các phiên bản trước như 0.8, trình sản xuất không đồng bộ không có lệnh gọi lại cho send () để đăng ký trình xử lý lỗi. Điều này chỉ có sẵn trong phiên bản 0.9 hiện tại.

public void close ()

Lớp nhà sản xuất cung cấp close phương pháp đóng các kết nối nhóm nhà sản xuất với tất cả Kafka bro-kers.

Thiết lập cấu hình

Cài đặt cấu hình chính của API Producer được liệt kê trong bảng sau để có tình trạng tốt hơn -

S.Không	Cài đặt cấu hình và mô tả
1	client.id xác định ứng dụng của nhà sản xuất
2	producer.type đồng bộ hoặc không đồng bộ
3	acks Cấu hình acks kiểm soát các tiêu chí theo yêu cầu của nhà sản xuất đã hoàn tất.
4	retries Nếu yêu cầu của nhà sản xuất không thành công, thì tự động thử lại với giá trị cụ thể.
5	bootstrap.servers bootstrapping danh sách các nhà môi giới.
6	linger.ms nếu bạn muốn giảm số lượng yêu cầu, bạn có thể đặt linger.ms thành giá trị lớn hơn một số giá trị.
7	key.serializer Phím cho giao diện bộ nối tiếp.
số 8	value.serializer giá trị cho giao diện bộ nối tiếp.
9	batch.size Kích thước đệm.
10	buffer.memory kiểm soát tổng dung lượng bộ nhớ có sẵn cho nhà sản xuất để buff-ering.

API ProducerRecord

ProducerRecord là một cặp khóa / giá trị được gửi đến phương thức khởi tạo lớp Kafka cluster.ProductionerRecord để tạo bản ghi với các cặp phân vùng, khóa và giá trị bằng cách sử dụng chữ ký sau.

public ProducerRecord (string topic, int partition, k key, v value)

Topic - tên chủ đề do người dùng xác định sẽ được thêm vào để ghi.
Partition - số lượng phân vùng
Key - Chìa khóa sẽ có trong hồ sơ.

Value - Ghi lại nội dung

public ProducerRecord (string topic, k key, v value)

Phương thức khởi tạo lớp ProducerRecord được sử dụng để tạo một bản ghi với các cặp khóa, giá trị và không có phân vùng.

Topic - Tạo chủ đề để gán bản ghi.
Key - chìa khóa cho bản ghi.
Value - ghi lại nội dung.

public ProducerRecord (string topic, v value)

Lớp ProducerRecord tạo một bản ghi không có phân vùng và khóa.

Topic - tạo chủ đề.
Value - ghi lại nội dung.

Các phương thức của lớp ProducerRecord được liệt kê trong bảng sau:

S.Không	Phương thức và mô tả lớp học
1	public string topic() Chủ đề sẽ thêm vào hồ sơ.
2	public K key() Chìa khóa sẽ được bao gồm trong hồ sơ. Nếu không có khóa như vậy, null sẽ được chuyển lại ở đây.
3	public V value() Ghi lại nội dung.
4	partition() Số lượng phân vùng cho bản ghi

Ứng dụng đơn giản

Trước khi tạo ứng dụng, đầu tiên hãy khởi động ZooKeeper và Kafka broker sau đó tạo chủ đề của riêng bạn trong Kafka broker bằng lệnh create topic. Sau đó, tạo một lớp java có tên Sim-pleProductioner.java và nhập mã sau.

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”
public class SimpleProducer {
   
   public static void main(String[] args) throws Exception{
      
      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }
      
      //Assign topicName to string variable
      String topicName = args[0].toString();
      
      // create instance for properties to access producer configs   
      Properties props = new Properties();
      
      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //Set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //Specify buffer size in config
      props.put("batch.size", 16384);
      
      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}

Compilation - Ứng dụng có thể được biên dịch bằng lệnh sau.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution - Ứng dụng có thể được thực thi bằng lệnh sau.

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Output

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

Ví dụ về người tiêu dùng đơn giản

Hiện tại, chúng tôi đã tạo một trình sản xuất để gửi tin nhắn đến Kafka cluster. Bây giờ chúng ta hãy tạo một người tiêu dùng để sử dụng các thông điệp tạo thành cụm Kafka. API KafkaConsumer được sử dụng để sử dụng các thông báo từ cụm Kafka. Phương thức khởi tạo lớp KafkaConsumer được định nghĩa bên dưới.

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

configs - Trả lại bản đồ cấu hình người tiêu dùng.

Lớp KafkaConsumer có các phương thức quan trọng sau đây được liệt kê trong bảng dưới đây.

S.Không	Phương pháp và Mô tả
1	public java.util.Set<TopicPar-tition> assignment() Nhận tập hợp các phân vùng hiện được chỉ định bởi con-sumer.
2	public string subscription() Đăng ký danh sách các chủ đề đã cho để nhận các phân vùng động như đã ký.
3	public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Đăng ký danh sách các chủ đề đã cho để nhận các phân vùng động như đã ký.
4	public void unsubscribe() Hủy đăng ký các chủ đề khỏi danh sách phân vùng đã cho.
5	public void sub-scribe(java.util.List<java.lang.String> topics) Đăng ký danh sách các chủ đề đã cho để nhận các phân vùng động như đã ký. Nếu danh sách các chủ đề nhất định trống, nó được coi giống như hủy đăng ký ().
6	public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) Mẫu đối số đề cập đến mẫu đăng ký ở định dạng biểu thức chính quy và đối số trình nghe nhận thông báo từ mẫu đăng ký.
7	public void as-sign(java.util.List<TopicParti-tion> partitions) Chỉ định thủ công danh sách các phân vùng cho khách hàng.
số 8	poll() Tìm nạp dữ liệu cho các chủ đề hoặc phân vùng được chỉ định bằng cách sử dụng một trong các API đăng ký / gán. Điều này sẽ trả về lỗi, nếu các chủ đề không được đăng ký trước khi bỏ phiếu cho dữ liệu.
9	public void commitSync() Các hiệu số cam kết được trả về trong cuộc thăm dò cuối cùng () cho tất cả danh sách chủ đề và phân vùng được ghi chép phụ. Thao tác tương tự cũng được áp dụng cho commitAsyn ().
10	public void seek(TopicPartition partition, long offset) Tìm nạp giá trị bù đắp hiện tại mà người tiêu dùng sẽ sử dụng trong phương thức thăm dò ý kiến () tiếp theo.
11	public void resume() Tiếp tục các phân vùng bị tạm dừng.
12	public void wakeup() Đánh thức người tiêu dùng.

API ConsumerRecord

API ConsumerRecord được sử dụng để nhận các bản ghi từ cụm Kafka. API này bao gồm tên chủ đề, số phân vùng mà bản ghi đang được nhận và phần bù trỏ đến bản ghi trong phân vùng Kafka. Lớp ConsumerRecord được sử dụng để tạo một bản ghi người tiêu dùng với tên chủ đề cụ thể, số lượng phân vùng và các cặp <key, value>. Nó có chữ ký sau.

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

Topic - Tên chủ đề cho hồ sơ người tiêu dùng nhận được từ cụm Kafka.
Partition - Phân vùng cho chủ đề.
Key - Khoá của bản ghi, nếu không có khoá nào tồn tại null sẽ được trả về.
Value - Ghi lại nội dung.

API ConsumerRecords

API ConsumerRecords hoạt động như một vùng chứa cho ConsumerRecord. API này được sử dụng để giữ danh sách ConsumerRecord trên mỗi phân vùng cho một chủ đề cụ thể. Constructor của nó được định nghĩa bên dưới.

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

TopicPartition - Trả lại bản đồ phân vùng cho một chủ đề cụ thể.

Records - Danh sách trả lại của ConsumerRecord.

Lớp ConsumerRecords có các phương thức sau được định nghĩa.

S.Không	Phương pháp và Mô tả
1	public int count() Số lượng bản ghi cho tất cả các chủ đề.
2	public Set partitions() Tập hợp các phân vùng có dữ liệu trong tập bản ghi này (nếu không có dữ liệu nào được trả về thì tập đó trống).
3	public Iterator iterator() Iterator cho phép bạn xoay vòng qua một tập hợp, thu thập hoặc di chuyển lại các phần tử.
4	public List records() Nhận danh sách các bản ghi cho phân vùng đã cho.

Thiết lập cấu hình

Cài đặt cấu hình cho cài đặt cấu hình chính API khách hàng tiêu dùng được liệt kê bên dưới -

S.Không	Cài đặt và Mô tả
1	bootstrap.servers Bootstrapping danh sách các nhà môi giới.
2	group.id Chỉ định một người tiêu dùng cá nhân vào một nhóm.
3	enable.auto.commit Bật cam kết tự động cho các giá trị bù nếu giá trị là true, nếu không thì không được cam kết.
4	auto.commit.interval.ms Trả lại tần suất các hiệu số đã tiêu thụ được cập nhật được ghi vào ZooKeeper.
5	session.timeout.ms Cho biết Kafka sẽ đợi bao nhiêu mili giây để ZooKeeper phản hồi một yêu cầu (đọc hoặc ghi) trước khi từ bỏ và tiếp tục sử dụng tin nhắn.

Ứng dụng SimpleConsumer

Các bước ứng dụng nhà sản xuất vẫn giữ nguyên ở đây. Đầu tiên, hãy khởi động nhà môi giới ZooKeeper và Kafka của bạn. Sau đó, tạo một ứng dụng SimpleConsumer với lớp java có tên SimpleCon-sumer.java và nhập mã sau.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();
      
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

Compilation - Ứng dụng có thể được biên dịch bằng lệnh sau.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − Ứng dụng có thể được thực thi bằng lệnh sau

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Input- Mở CLI của nhà sản xuất và gửi một số thông điệp cho chủ đề. Bạn có thể đặt đầu vào đơn giản là 'Xin chào người tiêu dùng'.

Output - Sau đây sẽ là đầu ra.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer

Nhóm người tiêu dùng là nhóm tiêu dùng đa luồng hoặc đa máy móc từ các chủ đề của Kafka.

Nhóm người tiêu dùng

Người tiêu dùng có thể tham gia một nhóm bằng cách sử dụng cùng một group.id.
Mức độ song song tối đa của một nhóm là số lượng người tiêu dùng trong nhóm ← không có phân vùng.
Kafka chỉ định các phân vùng của một chủ đề cho người tiêu dùng trong một nhóm, để mỗi phân vùng được sử dụng bởi chính xác một người tiêu dùng trong nhóm.
Kafka đảm bảo rằng một tin nhắn chỉ được đọc bởi một người tiêu dùng duy nhất trong nhóm.
Người tiêu dùng có thể xem thông báo theo thứ tự được lưu trong nhật ký.

Tái cân bằng của một người tiêu dùng

Việc thêm nhiều quy trình / luồng sẽ khiến Kafka cân bằng lại. Nếu bất kỳ người tiêu dùng hoặc nhà môi giới nào không gửi được nhịp tim tới ZooKeeper, thì nó có thể được cấu hình lại thông qua cụm Kafka. Trong quá trình cân bằng lại này, Kafka sẽ gán các phân vùng có sẵn cho các luồng có sẵn, có thể di chuyển một phân vùng sang một quy trình khác.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ConsumerGroup {
   public static void main(String[] args) throws Exception {
      if(args.length < 2){
         System.out.println("Usage: consumer <topic> <groupname>");
         return;
      }
      
      String topic = args[0].toString();
      String group = args[1].toString();
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", group);
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer",          
         "org.apache.kafka.common.serialization.ByteArraySerializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
      
      consumer.subscribe(Arrays.asList(topic));
      System.out.println("Subscribed to topic " + topic);
      int i = 0;
         
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
               System.out.printf("offset = %d, key = %s, value = %s\n", 
               record.offset(), record.key(), record.value());
      }     
   }  
}

Tổng hợp

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java

Chấp hành

>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group
>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group

Ở đây chúng tôi đã tạo một tên nhóm mẫu là nhóm của tôi với hai người tiêu dùng. Tương tự, bạn có thể tạo nhóm của mình và số lượng người tiêu dùng trong nhóm.

Đầu vào

Mở CLI của nhà sản xuất và gửi một số tin nhắn như -

Test consumer group 01
Test consumer group 02

Đầu ra của quy trình đầu tiên

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 01

Đầu ra của quy trình thứ hai

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 02

Bây giờ hy vọng bạn đã hiểu SimpleConsumer và ConsumeGroup bằng cách sử dụng bản trình diễn ứng dụng khách Java. Bây giờ bạn có ý tưởng về cách gửi và nhận tin nhắn bằng máy khách Java. Hãy để chúng tôi tiếp tục tích hợp Kafka với công nghệ dữ liệu lớn trong chương tiếp theo.

Trong chương này, chúng ta sẽ học cách tích hợp Kafka với Apache Storm.

Về Storm

Storm ban đầu được tạo ra bởi Nathan Marz và nhóm tại BackType. Trong một thời gian ngắn, Apache Storm đã trở thành một tiêu chuẩn cho hệ thống xử lý thời gian thực phân tán cho phép bạn xử lý một khối lượng dữ liệu khổng lồ. Storm rất nhanh và một điểm chuẩn đã ghi tốc độ của nó ở hơn một triệu bộ giá trị được xử lý mỗi giây trên mỗi nút. Apache Storm chạy liên tục, tiêu thụ dữ liệu từ các nguồn được cấu hình (Spouts) và chuyển dữ liệu xuống đường ống xử lý (Bu lông). Com-bined, Spouts và Bolts tạo nên một cấu trúc liên kết.

Tích hợp với Storm

Kafka và Storm bổ sung cho nhau một cách tự nhiên và sự hợp tác mạnh mẽ của họ cho phép phân tích phát trực tuyến theo thời gian thực cho dữ liệu lớn chuyển động nhanh. Tích hợp Kafka và Storm là để giúp các nhà phát triển nhập và xuất bản các luồng dữ liệu từ cấu trúc liên kết Storm dễ dàng hơn.

Dòng khái niệm

Vòi là một nguồn suối. Ví dụ: một vòi có thể đọc các bộ từ một Chủ đề Kafka và phát chúng dưới dạng một luồng. Một bu lông tiêu thụ các luồng đầu vào, xử lý và có thể phát ra các luồng mới. Bu lông có thể làm bất cứ điều gì từ chạy các chức năng, lọc bộ dữ liệu, tổng hợp phát trực tuyến, liên kết trực tuyến, nói chuyện với cơ sở dữ liệu, v.v. Mỗi nút trong cấu trúc liên kết Storm thực thi song song. Một cấu trúc liên kết chạy vô thời hạn cho đến khi bạn kết thúc nó. Storm sẽ tự động giao lại bất kỳ nhiệm vụ không thành công nào. Ngoài ra, Storm đảm bảo rằng sẽ không bị mất dữ liệu, ngay cả khi máy bị trục trặc và tin nhắn bị rớt.

Hãy cùng chúng tôi tìm hiểu chi tiết về API tích hợp Kafka-Storm. Có ba lớp chính để tích hợp Kafka với Storm. Chúng như sau:

BrokerHosts - ZkHosts & StaticHosts

BrokerHosts là một giao diện và ZkHosts và StaticHosts là hai cách triển khai chính của nó. ZkHosts được sử dụng để theo dõi động các nhà môi giới Kafka bằng cách duy trì các chi tiết trong ZooKeeper, trong khi StaticHosts được sử dụng để thiết lập thủ công / tĩnh các nhà môi giới Kafka và các chi tiết của nó. ZkHosts là cách đơn giản và nhanh chóng để truy cập nhà môi giới Kafka.

Chữ ký của ZkHosts như sau:

public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)

Trong đó brokerZkStr là máy chủ ZooKeeper và brokerZkPath là đường dẫn ZooKeeper để duy trì các chi tiết của nhà môi giới Kafka.

API KafkaConfig

API này được sử dụng để xác định cài đặt cấu hình cho cụm Kafka. Chữ ký của Kafka Con-fig được xác định như sau

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - Các BrokerHosts có thể là ZkHosts / StaticHosts.

Topic - tên chủ đề.

API SpoutConfig

Spoutconfig là một phần mở rộng của KafkaConfig hỗ trợ thêm thông tin ZooKeeper.

public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

Hosts - BrokerHosts có thể là bất kỳ triển khai nào của giao diện BrokerHosts
Topic - tên chủ đề.
zkRoot - Đường dẫn gốc ZooKeeper.
id −Vòi lưu trữ trạng thái của phần bù được tiêu thụ trong Zookeeper. Id sẽ xác định duy nhất vòi của bạn.

SchemeAsMultiScheme

SchemeAsMultiScheme là một giao diện chỉ định cách ByteBuffer tiêu thụ từ Kafka bị biến đổi thành một bộ ba cơn bão. Nó có nguồn gốc từ MultiScheme và chấp nhận triển khai lớp Scheme. Có rất nhiều cách triển khai của lớp Scheme và một trong những cách triển khai đó là StringScheme, phân tích cú pháp byte như một chuỗi đơn giản. Nó cũng kiểm soát việc đặt tên cho trường đầu ra của bạn. Chữ ký được định nghĩa như sau.

public SchemeAsMultiScheme(Scheme scheme)

Scheme - bộ đệm byte được sử dụng từ kafka.

API KafkaSpout

KafkaSpout là triển khai spout của chúng tôi, sẽ tích hợp với Storm. Nó lấy các mes-sage từ chủ đề kafka và phát nó vào hệ sinh thái Storm dưới dạng các bộ giá trị. KafkaSpout lấy chi tiết cấu hình của nó từ SpoutConfig.

Dưới đây là mã mẫu để tạo một vòi Kafka đơn giản.

// ZooKeeper connection string
BrokerHosts hosts = new ZkHosts(zkConnString);

//Creating SpoutConfig Object
SpoutConfig spoutConfig = new SpoutConfig(hosts, 
   topicName, "/" + topicName UUID.randomUUID().toString());

//convert the ByteBuffer to String.
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

//Assign SpoutConfig to KafkaSpout.
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

Tạo bu lông

Bolt là một thành phần lấy các bộ giá trị làm đầu vào, xử lý bộ giá trị và tạo ra các bộ giá trị mới làm đầu ra. Bu lông sẽ thực hiện giao diện IRichBolt. Trong chương trình này, hai lớp bu lông WordSplitter-Bolt và WordCounterBolt được sử dụng để thực hiện các hoạt động.

Giao diện IRichBolt có các phương thức sau:

Prepare- Cung cấp cho bu lông một môi trường để thực thi. Những người thực thi sẽ chạy phương thức này để khởi tạo vòi.
Execute - Xử lý một bộ dữ liệu đầu vào.
Cleanup - Được gọi khi có tia chớp tắt.
declareOutputFields - Khai báo lược đồ đầu ra của bộ tuple.

Hãy để chúng tôi tạo SplitBolt.java, thực hiện logic để tách một câu thành các từ và CountBolt.java, thực hiện logic để tách các từ duy nhất và đếm số lần xuất hiện của nó.

SplitBolt.java

import java.util.Map;

import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;

public class SplitBolt implements IRichBolt {
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
      OutputCollector collector) {
      this.collector = collector;
   }
   
   @Override
   public void execute(Tuple input) {
      String sentence = input.getString(0);
      String[] words = sentence.split(" ");
      
      for(String word: words) {
         word = word.trim();
         
         if(!word.isEmpty()) {
            word = word.toLowerCase();
            collector.emit(new Values(word));
         }
         
      }

      collector.ack(input);
   }
   
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }

   @Override
   public void cleanup() {}
   
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
   
}

CountBolt.java

import java.util.Map;
import java.util.HashMap;

import backtype.storm.tuple.Tuple;
import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;

public class CountBolt implements IRichBolt{
   Map<String, Integer> counters;
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
   OutputCollector collector) {
      this.counters = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple input) {
      String str = input.getString(0);
      
      if(!counters.containsKey(str)){
         counters.put(str, 1);
      }else {
         Integer c = counters.get(str) +1;
         counters.put(str, c);
      }
   
      collector.ack(input);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counters.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
   
   }

   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Trình lên cấu trúc liên kết

Cấu trúc liên kết Storm về cơ bản là một cấu trúc Tiết kiệm. Lớp TopologyBuilder cung cấp các phương thức đơn giản và dễ dàng để tạo các cấu trúc liên kết phức tạp. Lớp TopologyBuilder có các phương thức để thiết lập vòi (setSpout) và đặt chốt (setBolt). Cuối cùng, TopologyBuilder có createTopology để tạo thành-pology. shuffleGrouping và các phương thức fieldsGrouping giúp thiết lập nhóm luồng cho vòi và bu lông.

Local Cluster- Đối với mục đích phát triển, chúng ta có thể tạo ra một cụm địa phương sử dụng LocalCluster đối tượng và sau đó gửi topo sử dụng submitTopology phương pháp LocalCluster lớp.

KafkaStormSample.java

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.ZkHosts;
import storm.kafka.Broker;
import storm.kafka.StaticHosts;
import storm.kafka.BrokerHosts;
import storm.kafka.SpoutConfig;
import storm.kafka.KafkaConfig;
import storm.kafka.KafkaSpout;
import storm.kafka.StringScheme;

public class KafkaStormSample {
   public static void main(String[] args) throws Exception{
      Config config = new Config();
      config.setDebug(true);
      config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
      String zkConnString = "localhost:2181";
      String topic = "my-first-topic";
      BrokerHosts hosts = new ZkHosts(zkConnString);
      
      SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,    
         UUID.randomUUID().toString());
      kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.forceFromStart = true;
      kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));
      builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");
      builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");
         
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());

      Thread.sleep(10000);
      
      cluster.shutdown();
   }
}

Trước khi chuyển quá trình biên dịch, tích hợp Kakfa-Storm cần có thư viện java ứng dụng khách ZooKeeper của người phụ trách. Người quản lý phiên bản 2.9.1 hỗ trợ Apache Storm phiên bản 0.9.5 (chúng tôi sử dụng trong hướng dẫn này). Tải xuống các tệp jar được chỉ định bên dưới và đặt nó vào đường dẫn lớp java.

curator-client-2.9.1.jar
curator-framework-2.9.1.jar

Sau khi bao gồm các tệp phụ thuộc, hãy biên dịch chương trình bằng lệnh sau,

javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java

Chấp hành

Khởi động Kafka Producer CLI (đã giải thích ở chương trước), tạo một chủ đề mới được gọi là chủ đề đầu tiên của tôi và cung cấp một số thông báo mẫu như hình dưới đây -

hello
kafka
storm
spark
test message
another test message

Bây giờ thực thi ứng dụng bằng lệnh sau:

java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSample

Đầu ra mẫu của ứng dụng này được chỉ định bên dưới:

storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2

Trong chương này, chúng ta sẽ thảo luận về cách tích hợp Apache Kafka với Spark Streaming API.

Về Spark

Spark Streaming API cho phép xử lý luồng dữ liệu trực tiếp có thể mở rộng, thông lượng cao, chịu được lỗi. Dữ liệu có thể được nhập từ nhiều nguồn như Kafka, Flume, Twitter, v.v. và có thể được xử lý bằng các thuật toán phức tạp như các chức năng cấp cao như bản đồ, thu nhỏ, nối và cửa sổ. Cuối cùng, dữ liệu đã xử lý có thể được đẩy ra hệ thống tệp, cơ sở dữ liệu và bảng điều khiển trực tiếp. Tập dữ liệu phân tán có khả năng phục hồi (RDD) là một cấu trúc dữ liệu cơ bản của Spark. Nó là một tập hợp các đối tượng được phân phối bất biến. Mỗi tập dữ liệu trong RDD được chia thành các phân vùng logic, có thể được tính toán trên các nút khác nhau của cụm.

Tích hợp với Spark

Kafka là một nền tảng tích hợp và nhắn tin tiềm năng để phát trực tuyến Spark. Kafka đóng vai trò là trung tâm trung tâm cho các luồng dữ liệu thời gian thực và được xử lý bằng các thuật toán phức tạp trong Spark Streaming. Sau khi dữ liệu được xử lý, Spark Streaming có thể xuất bản kết quả vào một chủ đề Kafka khác hoặc lưu trữ trong HDFS, cơ sở dữ liệu hoặc trang tổng quan. Sơ đồ sau đây mô tả luồng khái niệm.

Bây giờ, hãy cùng chúng tôi tìm hiểu chi tiết về API Kafka-Spark.

API SparkConf

Nó đại diện cho cấu hình cho một ứng dụng Spark. Được sử dụng để đặt các tham số Spark khác nhau dưới dạng các cặp khóa-giá trị.

Lớp SparkConf có các phương thức sau:

set(string key, string value) - thiết lập biến cấu hình.
remove(string key) - xóa khóa khỏi cấu hình.
setAppName(string name) - đặt tên ứng dụng cho ứng dụng của bạn.
get(string key) - lấy chìa khóa

API StreamingContext

Đây là điểm nhập chính cho chức năng Spark. Một SparkContext đại diện cho kết nối với một cụm Spark và có thể được sử dụng để tạo RDD, bộ tích lũy và các biến quảng bá trên cụm. Chữ ký được xác định như hình bên dưới.

public StreamingContext(String master, String appName, Duration batchDuration, 
   String sparkHome, scala.collection.Seq<String> jars, 
   scala.collection.Map<String,String> environment)

master - URL cụm để kết nối (ví dụ: mesos: // host: port, spark: // host: port, local [4]).
appName - tên cho công việc của bạn, để hiển thị trên giao diện người dùng web cụm
batchDuration - khoảng thời gian mà dữ liệu truyền trực tuyến sẽ được chia thành các lô

public StreamingContext(SparkConf conf, Duration batchDuration)

Tạo StreamingContext bằng cách cung cấp cấu hình cần thiết cho SparkContext mới.

conf - Thông số tia lửa
batchDuration - khoảng thời gian mà dữ liệu truyền trực tuyến sẽ được chia thành các lô

API KafkaUtils

API KafkaUtils được sử dụng để kết nối cụm Kafka với luồng Spark. API này có chữ ký createStream của phương thức signifi -cant được định nghĩa như bên dưới.

public static ReceiverInputDStream<scala.Tuple2<String,String>> createStream(
   StreamingContext ssc, String zkQuorum, String groupId,
   scala.collection.immutable.Map<String,Object> topics, StorageLevel storageLevel)

Phương thức hiển thị ở trên được sử dụng để Tạo luồng đầu vào kéo các thông báo từ Kafka Brokers.

ssc - Đối tượng StreamingContext.
zkQuorum - túc số của Zookeeper.
groupId - Id nhóm cho người tiêu dùng này.
topics - trả lại một bản đồ các chủ đề để tiêu thụ.
storageLevel - Mức lưu trữ để sử dụng cho việc lưu trữ các đối tượng đã nhận.

KafkaUtils API có một phương thức khác là createDirectStream, được sử dụng để tạo luồng đầu vào trực tiếp kéo các thông báo từ Kafka Broker mà không cần sử dụng bất kỳ bộ thu nào. Luồng này có thể đảm bảo rằng mỗi tin nhắn từ Kafka được đưa vào các phép biến đổi chính xác một lần.

Ứng dụng mẫu được thực hiện trong Scala. Để biên dịch ứng dụng, vui lòng tải xuống và cài đặt công cụ xây dựng sbt , scala (tương tự như maven). Mã ứng dụng chính được trình bày bên dưới.

import java.util.HashMap

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

object KafkaWordCount {
   def main(args: Array[String]) {
      if (args.length < 4) {
         System.err.println("Usage: KafkaWordCount <zkQuorum><group> <topics> <numThreads>")
         System.exit(1)
      }

      val Array(zkQuorum, group, topics, numThreads) = args
      val sparkConf = new SparkConf().setAppName("KafkaWordCount")
      val ssc = new StreamingContext(sparkConf, Seconds(2))
      ssc.checkpoint("checkpoint")

      val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
      val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
      val words = lines.flatMap(_.split(" "))
      val wordCounts = words.map(x => (x, 1L))
         .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
      wordCounts.print()

      ssc.start()
      ssc.awaitTermination()
   }
}

Xây dựng tập lệnh

Việc tích hợp spark-kafka phụ thuộc vào tia lửa, dòng tia lửa và bình tích hợp tia lửa Kafka. Tạo một tệp build.sbt mới và chỉ định chi tiết ứng dụng và sự phụ thuộc của nó. Các SBT sẽ tải về jar cần thiết khi biên dịch và đóng gói ứng dụng.

name := "Spark Kafka Project"
version := "1.0"
scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"

Biên dịch / Đóng gói

Chạy lệnh sau để biên dịch và đóng gói tệp jar của ứng dụng. Chúng tôi cần gửi tệp jar vào bảng điều khiển tia lửa để chạy ứng dụng.

sbt package

Phục tùng Spark

Khởi động Kafka Producer CLI (đã giải thích ở chương trước), tạo một chủ đề mới được gọi là chủ đề đầu tiên của tôi và cung cấp một số thông báo mẫu như hình dưới đây.

Another spark test message

Chạy lệnh sau để gửi ứng dụng tới bảng điều khiển tia lửa.

/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class "KafkaWordCount" --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 <group name> <topic name> <number of threads>

Kết quả mẫu của ứng dụng này được hiển thị bên dưới.

spark console messages ..
(Test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..

Hãy để chúng tôi phân tích một ứng dụng theo thời gian thực để nhận các nguồn cấp dữ liệu twitter mới nhất và các thẻ bắt đầu bằng # của nó. Trước đó, chúng ta đã thấy sự tích hợp của Storm và Spark với Kafka. Trong cả hai tình huống, chúng tôi đã tạo Kafka Producer (sử dụng cli) để gửi thông điệp đến hệ sinh thái Kafka. Sau đó, tương tác giữa bão và tia lửa đọc các thông điệp bằng cách sử dụng người tiêu dùng Kafka và đưa nó vào hệ sinh thái bão và tia lửa tương ứng. Vì vậy, trên thực tế, chúng ta cần tạo một Kafka Producer, điều này sẽ -

Đọc các nguồn cấp dữ liệu twitter bằng cách sử dụng “Twitter Streaming API”,
Xử lý nguồn cấp dữ liệu,
Trích xuất HashTags và
Gửi nó cho Kafka.

Sau khi nhận được HashTags bởi Kafka, tích hợp Storm / Spark sẽ nhận thông tin và gửi nó đến hệ sinh thái Storm / Spark.

API phát trực tuyến Twitter

"Twitter Streaming API" có thể được truy cập bằng bất kỳ ngôn ngữ lập trình nào. “Twitter4j” là một thư viện Java không chính thức, mã nguồn mở, cung cấp một mô-đun dựa trên Java để dễ dàng truy cập vào “Twitter Streaming API”. “Twitter4j” cung cấp một khung dựa trên người nghe để truy cập các tweet. Để truy cập “API phát trực tuyến Twitter”, chúng tôi cần đăng nhập vào tài khoản nhà phát triển Twitter và sẽ nhận được các thông tin sauOAuth chi tiết xác thực.

Customerkey
CustomerSecret
AccessToken
AccessTookenSecret

Khi tài khoản nhà phát triển được tạo, hãy tải xuống tệp jar “twitter4j” và đặt nó vào đường dẫn lớp java.

Mã hóa nhà sản xuất Twitter Kafka hoàn chỉnh (KafkaTwitterProductioner.java) được liệt kê bên dưới:

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.*;
import twitter4j.conf.*;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaTwitterProducer {
   public static void main(String[] args) throws Exception {
      LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Sta-tus>(1000);
      
      if(args.length < 5){
         System.out.println(
            "Usage: KafkaTwitterProducer <twitter-consumer-key>
            <twitter-consumer-secret> <twitter-access-token>
            <twitter-access-token-secret>
            <topic-name> <twitter-search-keywords>");
         return;
      }
      
      String consumerKey = args[0].toString();
      String consumerSecret = args[1].toString();
      String accessToken = args[2].toString();
      String accessTokenSecret = args[3].toString();
      String topicName = args[4].toString();
      String[] arguments = args.clone();
      String[] keyWords = Arrays.copyOfRange(arguments, 5, arguments.length);

      ConfigurationBuilder cb = new ConfigurationBuilder();
      cb.setDebugEnabled(true)
         .setOAuthConsumerKey(consumerKey)
         .setOAuthConsumerSecret(consumerSecret)
         .setOAuthAccessToken(accessToken)
         .setOAuthAccessTokenSecret(accessTokenSecret);

      TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).get-Instance();
      StatusListener listener = new StatusListener() {
        
         @Override
         public void onStatus(Status status) {      
            queue.offer(status);

            // System.out.println("@" + status.getUser().getScreenName() 
               + " - " + status.getText());
            // System.out.println("@" + status.getUser().getScreen-Name());

            /*for(URLEntity urle : status.getURLEntities()) {
               System.out.println(urle.getDisplayURL());
            }*/

            /*for(HashtagEntity hashtage : status.getHashtagEntities()) {
               System.out.println(hashtage.getText());
            }*/
         }
         
         @Override
         public void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {
            // System.out.println("Got a status deletion notice id:" 
               + statusDeletionNotice.getStatusId());
         }
         
         @Override
         public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
            // System.out.println("Got track limitation notice:" + 
               num-berOfLimitedStatuses);
         }

         @Override
         public void onScrubGeo(long userId, long upToStatusId) {
            // System.out.println("Got scrub_geo event userId:" + userId + 
            "upToStatusId:" + upToStatusId);
         }      
         
         @Override
         public void onStallWarning(StallWarning warning) {
            // System.out.println("Got stall warning:" + warning);
         }
         
         @Override
         public void onException(Exception ex) {
            ex.printStackTrace();
         }
      };
      twitterStream.addListener(listener);
      
      FilterQuery query = new FilterQuery().track(keyWords);
      twitterStream.filter(query);

      Thread.sleep(5000);
      
      //Add Kafka producer config settings
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("acks", "all");
      props.put("retries", 0);
      props.put("batch.size", 16384);
      props.put("linger.ms", 1);
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer<String, String>(props);
      int i = 0;
      int j = 0;
      
      while(i < 10) {
         Status ret = queue.poll();
         
         if (ret == null) {
            Thread.sleep(100);
            i++;
         }else {
            for(HashtagEntity hashtage : ret.getHashtagEntities()) {
               System.out.println("Hashtag: " + hashtage.getText());
               producer.send(new ProducerRecord<String, String>(
                  top-icName, Integer.toString(j++), hashtage.getText()));
            }
         }
      }
      producer.close();
      Thread.sleep(5000);
      twitterStream.shutdown();
   }
}

Tổng hợp

Biên dịch ứng dụng bằng lệnh sau:

javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.java

Chấp hành

Mở hai bảng điều khiển. Chạy ứng dụng đã biên dịch ở trên như được hiển thị bên dưới trong một bảng điều khiển.

java -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:
. KafkaTwitterProducer <twitter-consumer-key>
<twitter-consumer-secret>
<twitter-access-token>
<twitter-ac-cess-token-secret>
my-first-topic food

Chạy bất kỳ ứng dụng Spark / Storm nào được giải thích trong chương trước trong một bản win-down khác. Điểm chính cần lưu ý là chủ đề được sử dụng phải giống nhau trong cả hai trường hợp. Ở đây, chúng tôi đã sử dụng “my-first-topic” làm tên chủ đề.

Đầu ra

Đầu ra của ứng dụng này sẽ phụ thuộc vào các từ khóa và nguồn cấp dữ liệu hiện tại của twitter. Đầu ra mẫu được chỉ định bên dưới (tích hợp bão).

. . .
food : 1
foodie : 2
burger : 1
. . .

Công cụ Kafka được đóng gói trong “org.apache.kafka.tools. *. Các công cụ được phân loại thành công cụ hệ thống và công cụ tái tạo.

Công cụ hệ thống

Các công cụ hệ thống có thể được chạy từ dòng lệnh bằng cách sử dụng tập lệnh lớp chạy. Cú pháp như sau:

bin/kafka-run-class.sh package.class - - options

Một số công cụ hệ thống được đề cập bên dưới:

Kafka Migration Tool - Công cụ này được sử dụng để di chuyển một nhà môi giới từ phiên bản này sang phiên bản khác.
Mirror Maker - Công cụ này được sử dụng để cung cấp phản chiếu của một cụm Kafka sang một cụm Kafka khác.
Consumer Offset Checker - Công cụ này hiển thị Nhóm người tiêu dùng, Chủ đề, Phân vùng, Off-set, logSize, Chủ sở hữu cho tập hợp Chủ đề và Nhóm người tiêu dùng được chỉ định.

Công cụ sao chép

Kafka replication là một công cụ thiết kế cấp cao. Mục đích của việc thêm công cụ sao chép là để có độ bền cao hơn và tính khả dụng cao hơn. Một số công cụ sao chép được đề cập dưới đây:

Create Topic Tool - Điều này tạo ra một chủ đề với số lượng phân vùng mặc định, hệ số nhân bản và sử dụng lược đồ mặc định của Kafka để thực hiện việc gán bản sao.
List Topic Tool- Công cụ này liệt kê thông tin cho một danh sách các chủ đề nhất định. Nếu không có chủ đề nào được cung cấp trong dòng lệnh, công cụ sẽ truy vấn Zookeeper để lấy tất cả các chủ đề và liệt kê thông tin cho chúng. Các trường mà công cụ hiển thị là tên chủ đề, phân vùng, thủ lĩnh, bản sao, isr.
Add Partition Tool- Tạo một chủ đề, số lượng phân vùng cho chủ đề phải được chỉ định. Sau đó, có thể cần nhiều phân vùng hơn cho chủ đề, khi đó khối lượng của chủ đề sẽ tăng lên. Công cụ này giúp thêm nhiều phân vùng cho một chủ đề cụ thể và cũng cho phép gán bản sao thủ công các phân vùng đã thêm.

Kafka hỗ trợ nhiều ứng dụng công nghiệp tốt nhất hiện nay. Chúng tôi sẽ cung cấp một cái nhìn tổng quan ngắn gọn về một số ứng dụng đáng chú ý nhất của Kafka trong chương này.

Twitter

Twitter là một dịch vụ mạng xã hội trực tuyến cung cấp nền tảng để gửi và nhận các tweet của người dùng. Người dùng đã đăng ký có thể đọc và đăng tweet, nhưng người dùng chưa đăng ký chỉ có thể đọc tweet. Twitter sử dụng Storm-Kafka như một phần của cơ sở hạ tầng xử lý luồng của họ.

Apache Kafka được sử dụng tại LinkedIn cho dữ liệu luồng hoạt động và số liệu hoạt động. Kafka mes-saging system giúp LinkedIn có nhiều sản phẩm khác nhau như LinkedIn Newsfeed, LinkedIn Today để tiêu thụ tin nhắn trực tuyến và ngoài các hệ thống phân tích ngoại tuyến như Hadoop. Độ bền mạnh mẽ của Kafka cũng là một trong những yếu tố quan trọng để kết nối với LinkedIn.

Netflix

Netflix là một nhà cung cấp đa quốc gia của Mỹ về phương tiện truyền phát trực tuyến Internet theo yêu cầu. Netflix sử dụng Kafka để theo dõi thời gian thực và xử lý sự kiện.

Mozilla

Mozilla là một cộng đồng phần mềm miễn phí, được thành lập vào năm 1998 bởi các thành viên của Netscape. Kafka sẽ sớm thay thế một phần của hệ thống sản xuất hiện tại của Mozilla để thu thập dữ liệu hiệu suất và sử dụng từ trình duyệt của người dùng cuối cho các dự án như Telemetry, Test Pilot, v.v.

Oracle

Oracle cung cấp kết nối gốc cho Kafka từ sản phẩm Bus dịch vụ doanh nghiệp có tên OSB (Oracle Service Bus) cho phép các nhà phát triển tận dụng khả năng dàn xếp tích hợp sẵn của OSB để triển khai các đường ống dữ liệu theo giai đoạn.