Streams และ Change Data Capture

การทำให้ Data Pipeline เป็นอัตโนมัติใน Snowflake

Emily Melhuish

Technical Curriculum Developer, Snowflake

ประมวลผลเฉพาะสิ่งที่เปลี่ยนแปลง

CDC: Change Data Capture การเปรียบเทียบสองแบบของการประมวลผลข้อมูล

Streams

หน้าที่ของ Stream

ติดตามทุก INSERT, UPDATE และ DELETE บนตารางต้นทาง

Screenshot 2026-05-11 at 10.50.31 am.png

รักษา change log ต่อเนื่อง — ไม่มีการทำซ้ำข้อมูล
เมื่อถูกใช้งาน offset จะเลื่อนไป; การอ่านครั้งถัดไปเริ่มใหม่

¹ * Snowflake Learning Resource

ประเภทของ Stream

ประเภท Stream	จับการเปลี่ยนแปลง	เหมาะสำหรับ
Standard	ตารางและวิวทุกประเภท & การเปลี่ยนแปลง DML ทั้งหมด - ติดตาม insert, update, delete	ตารางที่แถวใดก็ได้เปลี่ยนได้ (เช่น shipments)
Append-only	ตารางและวิวทุกประเภท ยกเว้น external table - ติดตามเฉพาะการ insert	ตารางที่ insert ครั้งเดียว (เช่น delivery events) - มีประสิทธิภาพกว่า
Insert-only	Apache Iceberg และ external table ที่จัดการภายนอก - ติดตามเฉพาะการ insert	External table

Directory table แสดง metadata ของไฟล์สำหรับ stage (ชื่อ, ขนาด, timestamp การแก้ไขล่าสุด)

การสร้าง Stream

Standard stream บนตาราง shipments

CREATE STREAM shipments_stream
  ON TABLE logistics.shipments;

Append-only stream บนตาราง delivery events

CREATE STREAM delivery_events_stream
  ON TABLE logistics.delivery_events
  APPEND_ONLY = TRUE;

คอลัมน์ Metadata ของ Stream

SELECT product, quantity, METADATA$ACTION, METADATA$ISUPDATE, METADATA$ROW_ID
FROM shipments_stream;

METADATA$ACTION: INSERT หรือ DELETE
METADATA$ISUPDATE: TRUE เมื่อเป็นส่วนหนึ่งของคู่การอัปเดต
METADATA$ROW_ID: ตัวระบุแถวทางกายภาพที่ไม่ซ้ำกัน
การอัปเดตปรากฏเป็นคู่ DELETE + INSERT โดยทั้งคู่ถูกตั้งค่า METADATA$ISUPDATE = TRUE

คอลัมน์ metadata ของ stream แสดง METADATA$ACTION, METADATA$ISUPDATE, METADATA$ROW_ID พร้อมข้อมูลตัวอย่าง

Stream Offset

แผนภาพไทม์ไลน์ - สร้าง Stream (offset เริ่มต้นที่นี่) → มีการเปลี่ยนแปลงในตารางต้นทาง (stream สะสมระเบียน) → Stream ถูกใช้งานในธุรกรรม (offset เลื่อนไปยังปัจจุบัน)

ภาพรวม Streams ใน Pipeline

Stream ทำงานร่วมกับ task — ออบเจกต์ Snowflake ที่รัน SQL ตามกำหนดเวลา
Task อ่านเฉพาะแถวที่เปลี่ยนแปลง; กับ 10 ล้านแถว: 2 วินาที เทียบกับ 2 นาที

Screenshot 2026-05-11 at 10.48.51 am.png

¹ * Snowflake Learning Resource

Query Streams ใน Pipeline

Stream ทำงานร่วมกับ task - ออบเจกต์ Snowflake ที่รัน SQL ตามกำหนดเวลา
Task อ่านเฉพาะแถวที่เปลี่ยนแปลง; กับ 10 ล้านแถว: 2 วินาที เทียบกับ 2 นาที

CREATE TASK logistics.sync_shipments
  WAREHOUSE = compute_wh
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('logistics.staging_shipments_stream')
AS
  INSERT INTO logistics.shipments
  SELECT shipment_id, region, carrier, delivery_days
  FROM logistics.staging_shipments_stream
  WHERE METADATA$ACTION = 'INSERT';

ลองฝึกกันเลย!

การทำให้ Data Pipeline เป็นอัตโนมัติใน Snowflake