Xây dựng pipeline dữ liệu trong Snowflake: Hướng dẫn thân thiện cho người mới bắt đầu

Pipeline dữ liệu là thành phần trung tâm của kỹ thuật dữ liệu. Chúng đại diện cho chuỗi bước đưa dữ liệu thô từ các hệ thống nguồn vào kho dữ liệu, nơi dữ liệu được làm sạch, chuyển đổi và sẵn sàng cho phân tích. Một pipeline được thiết kế tốt đảm bảo dữ liệu luân chuyển hiệu quả, đáng tin cậy và an toàn từ khâu thu nạp đến khâu tạo insight.

Snowflake, một nền tảng dữ liệu trên đám mây mạnh mẽ, đặc biệt phù hợp để xây dựng các pipeline dữ liệu có khả năng mở rộng, linh hoạt và tiết kiệm chi phí. Với các tính năng tích hợp sẵn cho việc nạp dữ liệu, chuyển đổi, tự động hóa và giám sát, Snowflake cho phép đội ngũ dữ liệu đơn giản hóa các quy trình phức tạp và tập trung vào việc tạo ra giá trị từ dữ liệu.

Trong hướng dẫn này, chúng ta sẽ đi qua quy trình xây dựng một pipeline dữ liệu trong Snowflake—from thiết lập môi trường đến tự động hóa các chuyển đổi dữ liệu bằng công cụ gốc như Streams và Tasks. Một số phần sẽ cần kiến thức về quy trình tích hợp và hệ thống quản lý sự kiện của dịch vụ đám mây. Với những phần này, tôi khuyến nghị bạn tham khảo tài liệu của dịch vụ đám mây mà bạn sử dụng.

Nếu bạn mới bắt đầu với Snowflake, hãy cân nhắc lộ trình kỹ năng Snowflake Foundations để thực hành trực tiếp và học cách truy vấn, mô hình hóa, cũng như quản lý dữ liệu. 

Tìm hiểu kiến trúc của Snowflake

Kiến trúc độc đáo của Snowflake được thiết kế rất tốt để hỗ trợ pipeline dữ liệu. Snowflake tách biệt compute và storage, nghĩa là bạn có thể mở rộng từng phần một cách độc lập. 

Dữ liệu được lưu trữ tập trung tại lớp lưu trữ đám mây, trong khi tính toán diễn ra tại các virtual warehouse, có thể khởi động, dừng và thay đổi kích thước khi cần.

Kiến trúc này cho phép nhiều người dùng và quy trình cùng truy cập dữ liệu mà không tranh chấp, khiến nó trở nên lý tưởng cho khối lượng công việc pipeline gồm thu nạp, chuyển đổi và phân tích. 

Nó cũng bổ sung một lớp an toàn, nơi các thay đổi dữ liệu có thể diễn ra mà không ảnh hưởng trực tiếp đến khả năng truy vấn dữ liệu của người dùng. 

Để biết thêm thông tin và tìm hiểu về lớp đám mây không được đề cập ở đây, tôi khuyến nghị bài viết đào sâu kiến trúc Snowflake.

Các thành phần chính cho pipeline Snowflake

Trong mỗi pipeline, chúng ta phải cân nhắc các bước cần thiết. Thông thường, bạn sẽ stage dữ liệu trước khi nạp vào Snowflake. Bản thân Snowflake xử lý việc lưu trữ dữ liệu, bao gồm các bảng và schema của chúng, trong khi virtual warehouse cung cấp tài nguyên tính toán cần thiết để thực thi truy vấn và các thao tác dữ liệu khác.

Dưới đây là định nghĩa rõ ràng về các thành phần thiết yếu được dùng trong pipeline dữ liệu Snowflake:

• Stage: Vị trí lưu trữ tạm thời hoặc bên ngoài dùng để tải vào/xuất dữ liệu.
• Virtual Warehouse: Chỉ cụ thể tài nguyên tính toán do Snowflake cung cấp. Virtual warehouse về bản chất là các cụm tính toán MPP (Massively Parallel Processing), được tạo thành từ nhiều máy ảo (VM). Snowflake cung cấp và quản lý các tài nguyên này để xử lý truy vấn và thực hiện chuyển đổi dữ liệu hiệu quả.
• Database: Vùng chứa logic để tổ chức dữ liệu. Thường lưu trữ một tập hợp các bảng.
• Schema: Phân khu bên trong một database để nhóm các bảng và đối tượng liên quan. Điều này định nghĩa cách các bảng liên hệ với nhau.
• Bảng: Nơi lưu trữ dữ liệu có cấu trúc.

Thiết lập môi trường Snowflake của bạn

Hãy đảm bảo bạn đã có sẵn một môi trường Snowflake sẵn sàng. Nếu bạn mới với Snowflake, hãy làm theo các bước sau để bắt đầu:

Tạo tài khoản Snowflake

Nếu bạn chưa có tài khoản Snowflake, có bản dùng thử 30 ngày bao gồm (tại thời điểm viết) $400 mức sử dụng miễn phí. Chừng đó là đủ để chúng ta thử vài pipeline mẫu!

1. Truy cập Snowflake Trial và đăng ký tài khoản miễn phí.

Biểu mẫu đăng ký cho Snowflake Trial

2. Cho mục đích của chúng ta, hãy chọn bản Standard thay vì Enterprise để tiết kiệm chi phí credit trong tương lai nếu bạn tiếp tục dùng Snowflake!
3. Chọn nhà cung cấp đám mây bạn muốn (AWS, Azure hoặc GCP). Ví dụ trong hướng dẫn này sẽ sử dụng AWS.
4. Trung tâm dữ liệu được chọn tự động nên là trung tâm gần bạn nhất về mặt địa lý và sẽ mang lại trải nghiệm tốt nhất.
5. Sau khi tạo tài khoản, đăng nhập vào Snowflake Web UI.

Điều hướng giao diện

Khi vào bên trong, bạn sẽ thấy:

• Worksheet: Nơi bạn có thể chạy các truy vấn SQL.
• Databases: Xem và quản lý database và schema.
• Warehouses: Giám sát và kiểm soát tài nguyên tính toán.
• History: Theo dõi các truy vấn và tác vụ đã chạy trước đó

Nếu bạn muốn có thêm lời khuyên về cách sử dụng Snowflake, hãy theo học khóa Snowflake Foundations này.

Tạo các tài nguyên cần thiết

Bước này sẽ bao gồm việc tạo virtual warehouse, database và schema. Chạy đoạn SQL sau trong worksheet của bạn để thiết lập môi trường:

-- set the user role appropriately
USE ROLE accountadmin;

-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;

-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;

-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;

-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;Được hỗ trợ bởi

Nạp dữ liệu vào Snowflake

Như chúng tôi thảo luận trong bài viết về Thu nạp dữ liệu trong Snowflake, Snowflake hỗ trợ cả nạp theo lô (bulk) và nạp liên tục. Nạp bulk được thiết kế để tải lên từng khối dữ liệu lớn vào bảng của bạn. Việc này thường được thực hiện theo lịch, hoặc tự động với cơ chế quản lý dựa trên sự kiện của Snowpipe. 

Nạp liên tục (dạng stream) có thể được thực hiện bằng các công cụ như Snowpipe Streaming. Để thu nạp dữ liệu, chúng ta phải tạo các tích hợp dữ liệu (stage) với nhà cung cấp đám mây và nạp dữ liệu bằng COPY INTO.

Tạo stage và integration với S3

Thu nạp dữ liệu là bước đầu tiên trong bất kỳ pipeline nào. Ở bước này, chúng ta sẽ đề cập các khái niệm cơ bản về tạo tích hợp lưu trữ và tạo stage. Để làm vậy, hãy giả định bạn đã kết nối Snowflake với AWS bằng một người dùng IAM của Snowflake. Nếu chưa, hãy tham khảo hướng dẫn cấu hình Snowflake  AWS trong tài liệu.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;Được hỗ trợ bởi

Cuối bước này, bạn sẽ tạo xong tích hợp Snowflake với AWS và tạo stage. Nếu bạn muốn ôn lại kiến thức cơ bản về AWS, bài viết sau là một giới thiệu tuyệt vời về AWS.

Nạp bulk với COPY INTO

Lệnh COPY INTO là phương pháp phổ biến để tải dữ liệu lên Snowflake. Chúng ta có thể dùng tệp tải lên cục bộ trong Snowflake, hoặc sử dụng kết nối đám mây. Trong ví dụ sau, chúng ta sẽ dùng stage để sao chép một tệp CSV.

-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Được hỗ trợ bởi

Tự động hóa việc nạp với Snowpipe

Việc nạp thủ công mỗi khi có tệp mới sẽ kém hiệu quả và tốn thời gian khi tốc độ tăng lên. Thay vào đó, chúng ta có thể dùng Snowpipe để nạp tệp từ stage. Sau đó, ta kết hợp thông báo sự kiện của AWS với người dùng IAM của Snowflake để kích hoạt Snowpipe mỗi lần có tệp mới.

-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Được hỗ trợ bởi

Tiếp theo, chúng ta có thể dùng các dịch vụ như Amazon SQS để gửi thông báo đến người dùng IAM của chúng ta bất cứ khi nào bucket đích nhận được tệp.

Cuối cùng, chúng ta có thể giám sát Snowpipe bằng:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;Được hỗ trợ bởi

Giờ đây bạn không còn phải lo tải dữ liệu mỗi khi có tệp đến bucket S3 nữa! Các thông báo đám mây sẽ kích hoạt Snowpipe để nạp dữ liệu. 

Chuyển đổi dữ liệu với Streams và Tasks

Streams và Tasks của Snowflake là các công cụ tích hợp để theo dõi và chuyển đổi dữ liệu thay đổi. Streams tập trung theo dõi thay đổi trên bảng, còn Tasks được dùng để tự động hóa việc thực thi câu lệnh SQL.

Giới thiệu về Streams

Stream thu thập thông tin CDC (Change Data Capture)—tức là insert, update và delete—trên một bảng kể từ lần truy vấn gần nhất. Những thay đổi này thường ở dạng các truy vấn DML (Data Manipulation Language). Điều này cho phép chúng ta thấy sự thay đổi của một bảng giữa hai thời điểm truy vấn.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Được hỗ trợ bởi

Sau đó, bạn có thể truy vấn stream để chỉ lấy các thay đổi:

SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';Được hỗ trợ bởi

Giới thiệu về Tasks

Tasks cho phép bạn tự động hóa việc thực thi SQL theo lịch hoặc khi có sự kiện kích hoạt. Tasks có thể gọi SQL, Javascript, Python, v.v., để tự động hóa các thủ tục trên bảng và pipeline của chúng ta.

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;Được hỗ trợ bởi

Kết hợp Streams và Tasks

Đây là nền tảng cho các pipeline chuyển đổi tự động. Như bạn thấy ở trên, Tasks có thể dùng Streams để kéo dữ liệu mới vào một bảng khác. Một quy trình tổng quát có thể như sau:

• Tạo một Stream để theo dõi thay đổi của bảng dữ liệu
• Tạo một Task lấy dữ liệu từ stream đó theo lịch định kỳ
• Mỗi lần lấy dữ liệu, chuyển đổi dữ liệu sang định dạng mong muốn
• Tải dữ liệu đó vào bảng đích

Bằng cách này, chúng ta đã tạo nên các phần của một pipeline dữ liệu. Kết hợp Streams và Tasks cho phép chúng ta chỉ chuyển đổi trên dữ liệu mới nạp. Nếu chỉ dùng Tasks, chúng ta có thể phải xử lý toàn bộ bảng raw mỗi lần muốn có dữ liệu mới—điều này có thể tốn kém và sử dụng tài nguyên tính toán kém hiệu quả.

Xây dựng một pipeline dữ liệu end-to-end

Hãy cùng đi qua một trường hợp sử dụng pipeline đơn giản: thu nạp dữ liệu đơn hàng, theo dõi thay đổi và chuyển đổi để phục vụ phân tích. Vì chúng ta đã trình bày cách thực hiện từng bước độc lập, phần này bao quát cách các bước trông như thế nào khi kết hợp với nhau. 

Bước 1: Tạo tích hợp lưu trữ đám mây và stage

Đầu tiên, chúng ta phải kết nối hệ sinh thái Snowflake với lưu trữ đám mây. Chúng ta dùng AWS làm ví dụ. Trong Snowflake, chúng ta dùng các lệnh CREATE STORAGE INTEGRATION và  CREATE STAGE để tạo kết nối giữa AWS và database của mình.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;Được hỗ trợ bởi

Bước 2: Thu nạp dữ liệu với Snowpipe

Ở bước hai, chúng ta tạo một Snowpipe sẽ chạy lệnh COPY INTO vào các bảng dữ liệu raw của chúng ta. 

-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS

COPY INTO datacamp_sample_db.raw_schema.aws_first_upload

FROM @datacamp_s3_stage

FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Được hỗ trợ bởi

Bước 3: Dùng Streams để bắt thay đổi

Sau đó, chúng ta dùng một stream trên bảng dữ liệu raw này để theo dõi thay đổi.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Được hỗ trợ bởi

Bước 4: Tự động hóa chuyển đổi với Tasks

Tiếp theo, chúng ta tạo một Task để chuyển đổi và nạp dữ liệu vừa nạp vào phiên bản đã xử lý của bảng. Task này sẽ kích hoạt theo lịch để kiểm tra dữ liệu mới. Nếu không có dữ liệu mới, Task sẽ không chạy. 

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;Được hỗ trợ bởi

Bước 5: Giám sát và quản lý

Cuối cùng, chúng ta muốn đảm bảo các pipeline hoạt động như mong đợi. Hãy dùng các lệnh SHOW PIPES và SHOW TASKS để xác minh rằng chúng được thiết lập và chạy đúng. Sau đó, dùng các bảng metadata từ INFORMATION_SCHEMA để theo dõi hiệu năng nạp dữ liệu và lỗi.

Thực hành tốt nhất khi phát triển pipeline dữ liệu trong Snowflake

Như thường lệ, chúng ta muốn đảm bảo đang vận hành theo các thực hành tốt nhất trong phát triển. Những điều cần cân nhắc gồm tối ưu cho hiệu năng và chi phí, đảm bảo chất lượng và độ tin cậy dữ liệu, cũng như triển khai các mức độ bảo mật và quyền truy cập phù hợp.

Tối ưu cho hiệu năng và chi phí

Mục tiêu của chúng ta nên là vận hành pipeline hiệu quả nhất với chi phí thấp nhất. Như vậy, ta giảm tác động lên máy chủ và cung cấp dữ liệu kịp thời nhất có thể. Dưới đây là một số mẹo: 

• Dùng auto-suspend cho warehouse để tránh phí nhàn rỗi. Điều này cắt giảm chi phí compute vì warehouse tính phí theo thời gian hoạt động.
• Tải các tệp đã nén và gom chúng thành kích thước tối ưu. Tệp quá nhỏ tiêu tốn quá nhiều chi phí mạng do phải tạo kết nối để tải từng tệp. Tệp quá lớn có thể chiếm quá nhiều băng thông và có nguy cơ lỗi. Snowflake khuyến nghị kích thước delta 100–250 MB.
• Lên lịch chuyển đổi trong các khung giờ ít sử dụng. Như vậy bạn không phải cạnh tranh tài nguyên.

Đảm bảo chất lượng và độ tin cậy dữ liệu

Chất lượng và độ tin cậy dữ liệu là yếu tố then chốt mà mọi kỹ sư dữ liệu phải cân nhắc. Bạn muốn đảm bảo dữ liệu được nạp có thể sử dụng mà không gặp lỗi. Bạn cũng muốn đảm bảo pipeline của mình sẽ không vô tình làm mất dữ liệu khi có sự cố. Dưới đây là vài mẹo quản lý chất lượng dữ liệu:

• Xác thực dữ liệu trước và sau khi thu nạp. Có thể thực hiện bằng các kiểm tra định kỳ trên bảng staging và thiết lập cảnh báo.
• Dùng streams và bảng staging để cô lập và sửa các bản ghi có vấn đề. Vì chỉ dữ liệu thay đổi mới được kéo vào, chúng ta có thể thu hẹp phạm vi vấn đề bằng cách dùng streams.
• Áp dụng ràng buộc duy nhất hoặc logic khử trùng lặp khi cần. Điều này có thể ngăn chặn dữ liệu NULL ở nơi không được phép có, hoặc khử trùng lặp trước trong trường hợp có các lượt kéo dữ liệu chồng lấn.

Triển khai bảo mật và kiểm soát truy cập

Cuối cùng, điều quan trọng là đảm bảo quyền truy cập và bảo mật phù hợp cho dữ liệu của bạn. Một số bảng có thể chứa thông tin nhạy cảm hoặc không nên bị thay đổi ngoại trừ bởi một nhóm chọn lọc. Hãy làm theo lời khuyên sau để quản lý bảo mật và quyền truy cập trong Snowflake một cách phù hợp:

• Sử dụng role và grant để hạn chế truy cập.
• Bảo vệ dữ liệu nhạy cảm bằng masking ở cấp cột hoặc chính sách truy cập theo hàng.
• Xoay vòng thông tin xác thực lưu trữ và giám sát việc sử dụng stage.

Kết luận

Xây dựng pipeline dữ liệu trong Snowflake là một quy trình gọn nhẹ và có khả năng mở rộng, rất phù hợp cho người mới bước vào thế giới kỹ thuật dữ liệu. Với kiến trúc trực quan, các tùy chọn thu nạp mạnh mẽ và công cụ tự động hóa tích hợp như Streams và Tasks, Snowflake giúp các đội ngũ dữ liệu nhanh chóng chuyển từ dữ liệu thô sang insight có thể hành động.

Bằng cách làm theo hướng dẫn này, giờ đây bạn đã có kiến thức nền tảng để thiết lập pipeline Snowflake của riêng mình—xử lý thu nạp, chuyển đổi và điều phối, tất cả trong một nền tảng duy nhất. Để biết thêm thông tin về Snowflake, hãy xem các hướng dẫn sau:

• Snowflake cho người mới bắt đầu
• Khóa học Giới thiệu về Snowflake
• Cách học Snowflake
• Mô hình dữ liệu trong Snowflake
• Chứng chỉ Snowflake tốt nhất 2025