Computer >> Hướng Dẫn Máy Tính >  >> Lập Trình >> Redis

Mô Hình Hóa Dữ Liệu Time Series Trong Redis Sử Dụng Streams

Streams: Cấu Trúc Dữ Liệu Mới Của Redis

Redis Streams đại diện cho một chuỗi thời gian chỉ thêm (append-only) của các cặp key-value.

Bất kỳ số lượng client nào cũng có thể ghi vào một Stream và mỗi khi họ làm như vậy, họ nhận được một ID duy nhất, luôn tăng dần của mục vừa được chèn vào chuỗi thời gian.

Client đọc dữ liệu có thể chặn (block) khi lắng nghe dữ liệu mới đến, có thể duy trì "đánh dấu" (bookmark) của tin nhắn cuối cùng đã đọc để xử lý theo lô (batch), hoặc có thể được tổ chức thành các "consumer groups" phức tạp hơn để chia sẻ khối lượng công việc và xác nhận (acknowledge) tin nhắn.

Streams là một chủ đề rộng, nên bài viết này chỉ xem xét sơ lược về cách sử dụng chúng để mô hình hóa dữ liệu time series trong Redis. Cấu trúc dữ liệu mới này làm cho việc này đơn giản hơn đáng kể so với các phương pháp trước đây như sử dụng List hoặc Sorted Set để mô hình hóa time series.

Đối với các ví dụ mã ở đây, chúng ta sẽ sử dụng client ioredis cho Node.js:

var Redis = require('ioredis'); var redis = new Redis();

Gửi Dữ Liệu Đến Stream

Streams là chỉ thêm (append-only), vì vậy thường thì bạn chỉ cần tên khóa Redis mà bạn muốn ghi vào và tập hợp các cặp key-value của mình.

Tôi sẽ ghi nhận các đo lường từ cảm biến chất lượng không khí của mình, gửi một luồng các cặp key-value đến khóa site:pdx - cụ thể là gửi Chỉ Số Chất Lượng Không Khí (AQI) hiện tại và nhiệt độ theo độ C:

redis.xadd('site:pdx', '*', 'aqi', 37, 'tempc', 5.1).then(function(id) { console.log("id:", id); });
> 1527974818120-0

Tôi gửi lệnh XADD mỗi khi có đo lường mới cần ghi lại. Phản hồi là một ID duy nhất, luôn tăng dần có thể được sử dụng trong các truy vấn. Phần đầu tiên của ID, 1527974818120 là dấu thời gian (timestamp) do máy chủ Redis gán. Phần thứ hai của ID là một số tăng dần để tránh va chạm khi nhiều client có thể ghi cùng một lúc.

Cung cấp * làm đối số thứ hai trong lệnh ghi, như trên, cho Redis biết lưu trữ dữ liệu với dấu thời gian của chính nó.

Cũng có thể cung cấp dấu thời gian khi ghi dữ liệu, nhưng thường không được ưu tiên: để Redis chọn dấu thời gian cho phép nhiều client ghi dữ liệu vào một Stream đồng thời mà không cần các client phối hợp về việc chọn và sắp xếp ID – máy chủ Redis chỉ xử lý các chi tiết đó cho các client.

Đọc Đơn Giản

Khi đã có một vài giá trị trong stream, bạn có thể truy xuất một phạm vi giá trị bằng cách cung cấp một phạm vi ID hoặc dấu thời gian. Điều này có thể được sử dụng bởi một ứng dụng hiển thị các đọc gần đây nhất của các chỉ số chất lượng không khí của tôi trên biểu đồ:

redis.xrange('site:pdx', '1527974818120-0', '+', 'COUNT', 5).then(function(resp) {
// resp hiện chứa 5 đọc, truyền chúng cho biểu đồ:
// console.log(resp);
});
> [
> [ '1543947167906-0', [ 'aqi', '31', 'tempc', '5.1' ] ],
> [ '1543947168312-0', [ 'aqi', '31', 'tempc', '5.3' ] ],
> [ '1543947168901-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947170033-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947171460-0', [ 'aqi', '31', 'tempc', '5.6' ] ]
> ]

Các phạm vi số có thể được lấy mẫu từ bất kỳ đâu trong Stream, cho phép hệ thống biểu đồ truy vấn dữ liệu lịch sử mà không bị phạt hiệu suất.

Chặn (Blocking) Và Polling

Truy vấn các phạm vi dữ liệu hữu ích cho biểu đồ và giám sát lịch sử, nhưng đôi khi bạn muốn xây dựng một hệ thống có thể phản ứng ngay lập tức với dữ liệu khi nó đến - Redis streams cũng rất phù hợp cho việc này, sử dụng XREAD:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', '$').then(function(resp) {
// đóng cửa sổ nếu aqi > 50
console.log(resp);
});

Một phép toán chặn như thế này sẽ chờ cho đến khi dữ liệu đến, hoặc cho đến khi hết thời gian chờ (10000ms ở đây), vì vậy để duy trì một poll liên tục, bạn chặn như trong ví dụ trên cho đến khi dữ liệu có sẵn, và đơn giản gọi lại lệnh XREAD đó mỗi lần sau khi nhận được dữ liệu hoặc lệnh hết thời gian chờ.

Để đảm bảo không bỏ lỡ dữ liệu giữa các kết nối đến stream của bạn, bạn có thể cung cấp ID cuối cùng đọc từ stream của bạn khi đọc từ nó, và như vậy bạn sẽ tiếp tục chính xác từ nơi bạn dừng lại.

Trong trường hợp trên, tôi chỉ muốn bất kỳ dữ liệu nào đến trên stream nên sử dụng token đặc biệt '$' để chỉ ra "chỉ dữ liệu mới"; lệnh trông chỉ khác một chút với ID được cung cấp:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', '1543947171460-0');

Bạn cũng có thể đọc qua nhiều stream khác nhau cùng một lúc, trả về giá trị từ stream đầu tiên chấp nhận dữ liệu:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', 'site:global', '1543947171460-0', '$');

Trong ví dụ này, lệnh sẽ trả về khi bất kỳ dữ liệu nào mới hơn ID 1543947171460-0 được ghi trên site:pdx, hoặc khi bất kỳ dữ liệu mới nào được ghi trên site:global.

Điều Phối Consumer

Khi bạn có nhiều dữ liệu chảy qua một stream, bạn có thể muốn có nhiều bản sao của consumer xử lý các tin nhắn đến. Các consumer này nên lấy một tin nhắn, thực hiện một hành động trên nó và sau đó "xác nhận" (acknowledge) rằng công việc đã được hoàn thành. Redis streams cung cấp các nguyên ngữ cho các hoạt động này cũng - quá nhiều để đi sâu vào đây nhưng hãy xem tài liệu để biết chi tiết.

Tra Cứu Nhanh Cho Các Lệnh Stream

Có rất nhiều lệnh Redis mới dành cho streams! Đây là một tra cứu nhanh, bạn có thể tìm thấy thêm chi tiết trong tài liệu chính thức.

Các lệnh đơn giản:

  • XADD: thêm một mục (một gói các cặp key-value) vào một stream
  • XRANGE và XREVRANGE: chọn phạm vi hoặc lặp lại các mục stream
  • XREAD: lấy các mục mới hơn một ID nào đó (tùy chọn chặn)
  • XTRIM: loại bỏ các mục cũ để cắt giảm stream
  • XDEL: xóa một mục cụ thể khỏi stream
  • XLEN: đếm số mục trong một stream
  • XINFO: kiểm tra metadata của stream

Các lệnh consumer group:

  • XGROUP: tạo, xóa, hoặc reset consumer groups, và xóa thành viên của chúng
  • XREADGROUP: giống XREAD (trên), nhưng nhận tin nhắn sử dụng consumer group
  • XPENDING: kiểm tra các tin nhắn đã được gửi đến consumer group nhưng chưa ack.
  • XACK: xác nhận một tin nhắn cho consumer group, xóa nó khỏi danh sách chờ
  • XCLAIM: lấy lại tin nhắn từ một consumer chết

Trường Hợp Sử Dụng

Trường hợp sử dụng được trích dẫn nhiều nhất cho Streams là các workload IoT nơi các cảm biến đặt dữ liệu lên một stream để tiêu thụ bởi các consumer sử dụng theo nhiều cách (phân tích, lưu trữ.archive vào cold storage, hiển thị trên biểu đồ). Streams với kích thước giới hạn là một trường hợp sử dụng tuyệt vời ở đây, cho phép bạn phân bổ một stream kích thước cố định với dấu vết bộ nhớ có thể dự đoán.

Streams cũng phù hợp để sử dụng trong các ứng dụng trước đây sử dụng các cấu trúc dữ liệu Redis khác. Các ứng dụng hàng đợi (queueing) như Celery và Sidekiq có thể tận dụng consumer groups của Streams để cung cấp kiểm tra read receipts theo cách Redis-native. Có nhiều bài blog trình diễn các ứng dụng chat đơn giản sử dụng Redis pubsub có thể được làm mạnh mẽ hơn với Redis Streams, vì pubsub không giữ lại tin nhắn sau khi chúng được phát hành đến các client.

Sẵn Sàng Thử?

Bạn có thể thử Redis streams trên một server RedisGreen mới chỉ với vài cú nhấp chuột.