การสร้างแบบจำลองข้อมูล Time Series ใน Redis ด้วย Streams

Redis มักถูกนึกถึงว่าเป็น "เซิร์ฟเวอร์โครงสร้างข้อมูล" ที่ให้อินเทอร์เฟซเครือข่ายสำหรับโครงสร้างข้อมูลพื้นฐานหลายประเภท Streams เป็นโครงสร้างข้อมูลแบบเจเนอริกใหม่ลำดับแรกนับตั้งแต่ Redis นำ Sorted Sets มาใช้งานเมื่อหลายปีก่อน มาดูหนึ่งในการใช้งานหลักของโครงสร้างใหม่นี้: การสร้างแบบจำลองข้อมูล time series

Streams: โครงสร้างข้อมูลใหม่ของ Redis

Redis Streams แสดงถึง time series แบบ append-only ของคู่คีย์-ค่า

ลูกค้าจำนวนไม่จำกัดสามารถเขียนข้อมูลลง Stream ได้ และทุกครั้งที่ทำเช่นนั้น จะได้รับ ID เฉพาะที่เพิ่มขึ้นเรื่อยๆ ของรายการที่ถูกแทรกเข้าไปใน time series

ลูกค้าที่อ่านข้อมูลสามารถบล็อกขณะรอข้อมูลใหม่เข้ามา รักษา "บุ๊กมาร์ก" ของข้อความล่าสุดที่อ่านเพื่อการประมวลผลแบบแบทช์ หรือจัดเป็น "consumer groups" ที่ซับซ้อนมากขึ้นเพื่อแบ่งปันภาระงานและยืนยันข้อความ

Streams เป็นหัวข้อกว้าง ข้อนี้จะเพียงมองสั้นๆ ว่าใช้สร้างแบบจำลอง time series ใน Redis ยังไง โครงสร้างข้อมูลใหม่นี้ทำให้เรื่องนี้ง่ายขึ้นมากกว่าวิธีเดิมอย่างมาก เช่น การใช้ List หรือ Sorted Set ในการสร้างแบบจำลอง time series

สำหรับตัวอย่างโค้ดที่นี่ เราจะใช้ไคลเอ็นต์ ioredis สำหรับ Node.js:

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

การส่งข้อมูลไปยัง Stream

Streams เป็นแบบ append-only ดังนั้นปกติแล้วคุณต้องการแค่ชื่อคีย์ Redis ที่ต้องการเขียน และชุดคู่คีย์-ค่าของคุณ

ผมจะบันทึกค่าจากเซ็นเซอร์คุณภาพอากาศ ส่ง stream ของคู่คีย์-ค่าไปที่คีย์ site:pdx โดยเฉพาะค่า Air Quality Index ปัจจุบันและอุณหภูมิในองศาเซลเซียส:

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

ผมส่งคำสั่ง XADD ทุกครั้งที่มีการวัดค่าใหม่ ตอบกลับเป็น ID เฉพาะที่เพิ่มขึ้นเรื่อยๆ สามารถใช้ในการคิวรี ส่วนแรกของ ID คือ 1527974818120 เป็น timestamp ที่ Redis server กำหนด ส่วนที่สองเป็นตัวเลขเพิ่มขึ้นเพื่อกันการชนกันเมื่อหลายไคลเอ็นต์เขียนพร้อมกัน

การส่ง * เป็นอาร์กิวเมนต์ที่สองในคำสั่งเขียน จะบอก Redis ให้เก็บข้อมูลพร้อม timestamp ของตัวมันเอง

ยังสามารถระบุ timestamp เองขณะเขียนได้ แต่ไม่แนะนำ: ให้ Redis เลือก timestamp เอง จะทำให้หลายไคลเอ็นต์เขียนข้อมูลลง Stream เดียวกันพร้อมกันได้โดยไม่ต้องประสานงานเรื่องการเลือกและเรียงลำดับ ID – Redis server จะจัดการรายละเอียดเหล่านี้ให้ไคลเอ็นต์

การอ่านแบบง่าย

เมื่อมีค่าบางส่วนอยู่ใน stream แล้ว คุณสามารถดึงช่วงค่าได้โดยระบุช่วง ID หรือ timestamp นี่ใช้ได้กับแอปที่แสดงค่าล่าสุดของเมตริกคุณภาพอากาศบนกราฟ:

redis.xrange('site:pdx', '1527974818120-0', '+', 'COUNT', 5).then(function(resp) { // resp ตอนนี้มี 5 ค่าการอ่าน ส่งต่อให้กราฟ: // 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' ] ] ]

ช่วงตัวเลขสามารถสุ่มตัวอย่างได้จากที่ไหนใน Stream ก็ได้ ทำให้ระบบกราฟสามารถคิวรีข้อมูลย้อนหลังได้โดยไม่มีค่าปรับด้านประสิทธิภาพ

การบล็อกและการโพลลิ่ง

การคิวรีช่วงข้อมูลมีประโยชน์สำหรับกราฟและการตรวจสอบย้อนหลัง แต่บางครั้งคุณอยากสร้างระบบที่ตอบสนองทันทีต่อข้อมูลที่เข้ามา – Redis streams เหมาะสำหรับกรณีนี้เช่นกัน โดยใช้ XREAD:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', '$').then(function(resp) { // ปิดหน้าต่างถ้า aqi > 50 console.log(resp); });

การดำเนินการบล็อกแบบนี้จะรอจนกว่าจะมีข้อมูลเข้ามา หรือจนถึง timeout (1000ms ที่นี่) เพื่อรักษาการโพลลิ่งต่อเนื่อง ให้บล็อกเหมือนตัวอย่างข้างต้นจนกว่าจะมีข้อมูล และเรียกคำสั่ง XREAD เดียวกันซ้ำทุกครั้งหลังจากได้ข้อมูลหรือคำสั่งหมดเวลา

เพื่อให้แน่ใจว่าไม่พลาดข้อมูลระหว่างการเชื่อมต่อกับ stream คุณสามารถส่ง ID ล่าสุดที่อ่านจาก stream ของคุณเมื่ออ่าน และด้วยวิธีนี้คุณจะเริ่มต่อจากจุดที่หยุดไว้ได้พอดี

ในกรณีข้างต้นผมแค่อยากได้ข้อมูลใหม่ที่เข้ามาใน stream จึงใช้โทเคนพิเศษ '$' เพื่อระบุว่า "ข้อมูลใหม่เท่านั้น" คำสั่งจะต่างออกไปเพียงเล็กน้อยเมื่อมีการระบุ ID:

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

คุณยังอ่านข้าม stream หลายตัวพร้อมกันได้ คืนค่าจาก stream แรกที่รับข้อมูล:

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

ในตัวอย่างนี้ คำสั่งจะคืนค่าเมื่อมีข้อมูลใหม่กว่า ID 1543947171460-0 ถูกเขียนบน site:pdx หรือเมื่อมีข้อมูลใหม่ใดๆ ถูกเขียนบน site:global

การประสานงาน Consumer

เมื่อมีข้อมูลไหลผ่าน stream มากขึ้น คุณอาจอยากให้มีหลายสำเนาของ consumer ประมวลผลข้อความเข้า Consumer ควรหยิบข้อความ ดำเนินการ และ "ยืนยัน" (acknowledge) ว่างานเสร็จแล้ว Redis streams ให้ primitive สำหรับการดำเนินการเหล่านี้เช่นกัน – รายละเอียดเยอะเกินไปที่จะอธิบายที่นี่ ดูรายละเอียดในเอกสารอย่างเป็นทางการ

คู่มืออ้างอิงเร็วสำหรับคำสั่ง Stream

มีคำสั่ง Redis ใหม่เยอะมากสำหรับ streams! นี่คือคู่มืออ้างอิงเร็ว รายละเอียดเพิ่มเติมในเอกสารอย่างเป็นทางการ

คำสั่งพื้นฐาน:

• XADD: เพิ่มรายการ (ชุดคู่คีย์-ค่า) ลงใน stream
• XRANGE และ XREVRANGE: เลือกช่วงหรือวนลูปผ่านรายการใน stream
• XREAD: ดึงรายการที่ใหม่กว่า ID บางตัว (เลือกบล็อกได้)
• XTRIM: ตัดรายการเก่าออกเพื่อปรับขนาด stream
• XDEL: ลบรายการเฉพาะจาก stream
• XLEN: นับจำนวนรายการใน stream
• XINFO: ตรวจสอบ metadata ของ stream

คำสั่ง Consumer Group:

• XGROUP: สร้าง ลบ หรือรีเซ็ต consumer groups และลบสมาชิก
• XREADGROUP: เหมือน XREAD (ด้านบน) แต่รับข้อความผ่าน consumer group
• XPENDING: ตรวจสอบข้อความที่ส่งไปยัง consumer group แล้วแต่ยังไม่ได้ ack
• XACK: ยืนยันข้อความสำหรับ consumer group ลบออกจากรายการ pending
• XCLAIM: รับช่วงข้อความจาก consumer ที่ตาย

กรณีการใช้งาน (Use Cases)

กรณีใช้งานที่พบบ่อยที่สุดสำหรับ Streams คือภาระงาน IoT ที่เซ็นเซอร์วางข้อมูลลง stream เพื่อให้ consumer ใช้งานหลายทาง (วิเคราะห์ จัดเก็บแบบ cold storage แสดงบนกราฟ) Streams ที่จำกัดขนาดเป็นกรณีใช้งานดีที่นี่ อนุญาตให้จัดสรร stream ขนาดคงที่ด้วย memory footprint ที่คาดเดาได้

Streams ยังเหมาะสำหรับแอปที่เคยใช้โครงสร้างข้อมูล Redis อื่นๆ แอปคิวเหมือน Celery และ Sidekiq สามารถใช้ประโยชน์จาก consumer groups ของ Streams เพื่อให้การตรวจสอบ receipt การอ่านแบบ Redis-native มีบล็อกโพสต์สาธิตแอปแชทง่ายๆ ใช้ Redis pubsub ซึ่งสามารถทำให้แข็งแกร่งกว่าด้วย Redis Streams เนื่องจาก pubsub ไม่เก็บข้อความหลังเผยแพร่ไปยังไคลเอ็นต์แล้ว

พร้อมลองใช้ไหม?

คุณสามารถลอง Redis streams บนเซิร์ฟเวอร์ RedisGreen ใหม่ได้เพียงไม่กี่คลิก