Xây dựng Pipeline Dữ liệu bằng R

Học cách biến đổi và nạp (ETL) một pipeline dữ liệu từ đầu bằng R và SQLite để thu thập tweet theo thời gian thực và lưu trữ cho các phân tích sau này.

Đã cập nhật 5 thg 10, 2026 · 13 phút đọc

Khám phá cùng AI

ChatGPT Claude Perplexity

Có thể lập luận rằng các pipeline ETL đúng chuẩn là một bộ phận sống còn của khoa học dữ liệu. Không có dữ liệu sạch và được tổ chức, sẽ rất khó tạo ra những insight chất lượng để nâng cao quyết định kinh doanh.

Vì vậy, trong hướng dẫn này, chúng ta sẽ khám phá cách xây dựng một pipeline ETL đơn giản để truyền tweet theo thời gian thực trực tiếp vào một cơ sở dữ liệu SQLite bằng R. Đây là một tác vụ khá phổ biến trong phân tích mạng xã hội, chẳng hạn.

Trọng tâm sẽ là trình bày tư duy về thu thập và lưu trữ dữ liệu cũng như cách vận hành Twitter API bằng gói rtweet trong R.

Để bắt đầu, hãy chuẩn bị đầy đủ công cụ cần thiết. Việc đầu tiên bạn cần là thiết lập quyền truy cập Twitter API. Nhìn chung, bạn cần thực hiện các bước sau:

Tạo tài khoản Twitter, nếu bạn chưa có.
Làm theo liên kết này và đăng ký tài khoản nhà phát triển (lưu ý quy trình này hiện yêu cầu đơn đăng ký và Twitter phải phê duyệt).
Tạo một ứng dụng mới tại trang web sau.
Điền mọi thông tin về ứng dụng của bạn và tạo access token.
Thu thập consumer key, consumer secret, access token và access token secret, là các thông tin bạn cần để kết nối với API.

Khi đã thiết lập xong Twitter API, bạn cần cài SQLite nếu chưa có. Để xem đầy đủ quy trình cài SQLite trên máy, hãy theo dõi hướng dẫn Beginners Guide to SQLite tại DataCamp. Việc chọn SQLite cho hướng dẫn này là do công cụ này rất đơn giản khi vận hành.

Bước 1: Tạo Cơ sở Dữ liệu và Bảng để Lưu Trữ Dữ liệu Twitter

Khi đã có quyền truy cập Twitter API và cài đặt SQLite, chúng ta có thể bắt đầu xây dựng pipeline để lưu trữ tweet khi chúng được truyền theo thời gian. Trước tiên, chúng ta sẽ tạo cơ sở dữ liệu SQLite mới bằng R như sau:

# Import necessary libraries and functions
library(RSQLite)
library(rtweet)
library(tm)
library(dplyr)
library(knitr)
library(wordcloud)
library(lubridate)
library(ggplot2)
source("transform_and_clean_tweets.R")

# Create our SQLite database
conn <- dbConnect(RSQLite::SQLite(), "Tweet_DB.db")

Sau đó, chúng ta có thể tạo một bảng bên trong cơ sở dữ liệu để chứa các tweet. Trong trường hợp cụ thể này, chúng ta sẽ lưu trữ các biến sau:

Tweet_ID là khóa chính INTEGER
User là TEXT
Tweet_Content là TEXT
Date_Created là INTEGER

Bạn có thể thắc mắc: tại sao tôi đặt ngày dưới dạng số nguyên? Vì SQLite không có kiểu dữ liệu dành riêng cho ngày giờ. Do đó, ngày sẽ được lưu dưới dạng số giây kể từ 1970-01-01.

Giờ hãy viết câu lệnh tạo bảng:

dbExecute(conn, "CREATE TABLE Tweet_Data(
                  Tweet_ID INTEGER PRIMARY KEY,
                  User TEXT,
                  Tweet_Content TEXT,
                  Date_Created INTEGER)")

Sau khi tạo bảng, bạn có thể mở sqlite3.exe và kiểm tra để chắc rằng bảng đã được tạo. Ảnh chụp màn hình minh họa bên dưới:

Bước 2: Truyền Tweet về Chủ đề Bạn Ưa Thích!

Nghe có vẻ khó tin, nhưng bạn đã hoàn tất về mặt yêu cầu và hạ tầng để có một pipeline truyền tweet đơn giản, hoạt động được. Việc cần làm bây giờ là truyền tweet bằng API. Lưu ý rằng cho mục đích hướng dẫn này, tôi sẽ dùng API miễn phí tiêu chuẩn. Có các phiên bản trả phí cao cấp có thể phù hợp hơn với nhu cầu truyền tweet của bạn nếu bạn thực hiện nghiên cứu, chẳng hạn.

Không chần chừ nữa, hãy bắt đầu thiết lập bộ lắng nghe Twitter. Việc đầu tiên là nhập gói rtweet và nhập các access token và secret của ứng dụng của bạn như đã mô tả ở phần đầu:

token <- create_token(app = 'Your_App_Name',
                      consumer_key = 'Your_Consumer_Key',
                      consumer_secret = 'Your_Consumer_Secret',
                      access_token = 'Your_Access_Token',
                      access_secret = 'Your_Access_Secret')

Khi đã có token, bước tiếp theo là quyết định bạn muốn truyền những tweet nào (tức là lắng nghe gì). Hàm stream_tweets trong gói rtweet cung cấp nhiều tùy chọn để truy vấn Twitter API. Ví dụ, bạn có thể truyền các tweet chứa một hoặc nhiều hashtag hay từ khóa nhất định (tối đa 400), một tập con ngẫu nhiên nhỏ của mọi tweet công khai, theo dõi tweet của một nhóm ID người dùng hoặc tên màn hình (tối đa 5000), hoặc thu thập tweet theo vị trí địa lý.

Trong hướng dẫn này, tôi quyết định truyền các tweet chứa hashtag liên quan đến khoa học dữ liệu (xem danh sách bên dưới). Bạn có thể nhận thấy định dạng tôi dùng để khai báo các hashtag có vẻ hơi lạ. Tuy nhiên, đó là định dạng mà hàm stream_tweets yêu cầu khi bạn muốn lắng nghe hashtag hoặc từ khóa. Định dạng này sẽ khác nếu bạn định lắng nghe một tập người dùng nhất định hoặc dựa trên tọa độ. Để biết thêm chi tiết, hãy tham khảo tài liệu.

keys <- "#nlp,#machinelearning,#datascience,#chatbots,#naturallanguageprocessing,#deeplearning"

Sau khi xác định từ khóa, đến lúc định nghĩa vòng lặp truyền tweet. Có nhiều cách làm việc này, nhưng định dạng dưới đây đã hoạt động tốt với tôi trước đây:

# Initialize the streaming hour tally
hour_counter <- 0

# Initialize a while loop that stops when the number of hours you want to stream tweets for is exceeded
while(hour_counter <= 12){
  # Set the stream time to 2 hours each iteration (7200 seconds)
  streamtime <- 7200
  # Create the file name where the 2 hour stream will be stored. Note that the Twitter API outputs a .json file.
  filename <- paste0("nlp_stream_",format(Sys.time(),'%d_%m_%Y__%H_%M_%S'),".json")
  # Stream Tweets containing the desired keys for the specified amount of time
  stream_tweets(q = keys, timeout = streamtime, file_name = filename)
  # Clean the streamed tweets and select the desired fields
  clean_stream <- transform_and_clean_tweets(filename, remove_rts = TRUE)
  # Append the streamed tweets to the Tweet_Data table in the SQLite database
  dbWriteTable(conn, "Tweet_Data", clean_stream, append = T)
  # Delete the .json file from this 2-hour stream
  file.remove(filename)
  # Add the hours to the tally
  hour_counter <- hour_counter + 2
}

Về bản chất, vòng lặp đó truyền càng nhiều tweet càng tốt có nhắc đến bất kỳ hashtag nào trong chuỗi từ khóa, theo từng khoảng 2 giờ trong tổng thời gian 12 giờ. Cứ mỗi 2 giờ, bộ lắng nghe Twitter tạo một tệp .json trong thư mục làm việc hiện tại với tên được chỉ định trong biến filename.

Sau đó, tệp này được truyền vào hàm transform_and_clean_tweets để loại bỏ retweet nếu muốn, chọn các cột chúng ta cần giữ từ tất cả những cột do Twitter API cung cấp, và chuẩn hóa văn bản trong các tweet.

Tiếp theo, nó nối dataframe kết quả vào bảng Tweet_Data mà ta đã tạo trong cơ sở dữ liệu SQLite. Cuối cùng, nó cộng thêm 2 vào biến đếm giờ (vì mỗi lần truyền kéo dài 2 giờ), và xóa tệp .json đã tạo. Việc này được thực hiện vì toàn bộ dữ liệu quan tâm đã nằm trong cơ sở dữ liệu, còn giữ lại các tệp .json có thể trở thành gánh nặng lưu trữ.

Giờ hãy xem kỹ hơn hàm transform_and_clean_tweets:

transform_and_clean_tweets <- function(filename, remove_rts = TRUE){

  # Import the normalize_text function
  source("normalize_text.R")

  # Parse the .json file given by the Twitter API into an R data frame
  df <- parse_stream(filename)
  # If remove_rst = TRUE, filter out all the retweets from the stream
  if(remove_rts == TRUE){
    df <- filter(df,df$is_retweet == FALSE)
  }
  # Keep only the tweets that are in English
  df <- filter(df, df$lang == "en")
  # Select the features that you want to keep from the Twitter stream and rename them
  # so the names match those of the columns in the Tweet_Data table in our database
  small_df <- df[,c("screen_name","text","created_at")]
  names(small_df) <- c("User","Tweet_Content","Date_Created")
  # Finally normalize the tweet text
  small_df$Tweet_Content <- sapply(small_df$Tweet_Content, normalize_text)
  # Return the processed data frame
  return(small_df)
}

Như đã lưu ý, hàm này dùng để lọc retweet nếu muốn, giữ lại các đặc trưng cần thiết và chuẩn hóa văn bản tweet. Về cơ bản, đây có thể xem là phần "T" trong chữ viết tắt ETL (transformation). Một thành phần quan trọng của hàm là quá trình làm sạch văn bản tweet.

Thông thường, dữ liệu văn bản cần một vài bước tiền xử lý trước khi sẵn sàng cho bất kỳ loại phân tích nào. Với tweet, các bước này có thể bao gồm loại bỏ URL, stop word và lượt đề cập, chuyển văn bản thành chữ thường, stemming, v.v. Tuy nhiên, không phải lúc nào tất cả bước này cũng cần thiết. Dù vậy, tạm thời tôi sẽ trình bày hàm normalize_text mà tôi dùng để tiền xử lý các tweet này dưới đây:

normalize_text <- function(text){
  # Keep only ASCII characters
  text = iconv(text, "latin1", "ASCII", sub="")
  # Convert to lower case characters
  text = tolower(text)
  # Remove any HTML tags
  text = gsub("<.*?>", " ", text)
  # Remove URLs
  text = gsub("\\s?(f|ht)(tp)(s?)(://)([^\\.]*)[\\.|/](\\S*)", "", text)
  # Keep letters and numbers only
  text = gsub("[^[:alnum:]]", " ", text)
  # Remove stop words
  text = removeWords(text,c("rt","gt",stopwords("en")))
  # Remove any extra white space
  text = stripWhitespace(text)                                 
  text = gsub("^\\s+|\\s+$", "", text)                         

  return(text)
}

Tùy trường hợp sử dụng, những bước này có thể là đủ. Như đã đề cập, bạn có thể thêm các bước khác như stemming hay lemmatization, hoặc chỉ giữ chữ cái thay vì cả chữ và số. Tôi khuyến khích bạn thử nghiệm và thử các kết hợp khác nhau. Đây cũng là nơi tuyệt vời để luyện kỹ năng regex.

Sau tất cả các bước trên, kết quả là một cơ sở dữ liệu SQLite đã được điền đầy các tweet đã truyền. Bạn có thể kiểm tra mọi thứ hoạt động đúng bằng cách chạy vài truy vấn đơn giản, ví dụ:

data_test <- dbGetQuery(conn, "SELECT * FROM Tweet_Data LIMIT 20")
unique_rows <- dbGetQuery(conn, "SELECT COUNT() AS Total FROM Tweet_Data")
kable(data_test)

Cơ sở dữ liệu SQLite được điền đầy các tweet đã truyền

print(as.numeric(unique_rows))

## [1] 1863

Bước 3: Phân tích

Khi đã chắc chắn quy trình ETL hoạt động như mong muốn, bước cuối cùng là thu thập một số insight và phân tích dữ liệu bạn đã thu thập. Ví dụ, với các tweet chúng ta có, ta có thể thử vài việc khác nhau: wordcloud về các thuật ngữ được đề cập trong nội dung tweet và một biểu đồ dòng thời gian để trực quan hóa trong khoảng 12 giờ truyền, ta thu được nhiều tweet nhất vào thời điểm nào. Dĩ nhiên đây không phải là danh sách đầy đủ mọi dạng nghiên cứu có thể thực hiện với tweet, từ phân tích cảm xúc đến nghiên cứu tâm lý học và hơn thế nữa.

Nói vậy là đủ, hãy bắt tay ngay vào dựng một wordcloud đẹp mắt:

# Gather all tweets from the database
all_tweets <- dbGetQuery(conn, "SELECT Tweet_ID, Tweet_Content FROM Tweet_Data")

# Create a term-document matrix and sort the words by frequency
dtm <- TermDocumentMatrix(VCorpus(VectorSource(all_tweets$Tweet_Content)))
dtm_mat <- as.matrix(dtm)
sorted <- sort(rowSums(dtm_mat), decreasing = TRUE)
freq_df <- data.frame(words = names(sorted), freq = sorted)

# Plot the wordcloud
set.seed(42)
wordcloud(words = freq_df$words, freq = freq_df$freq, min.freq = 10,
          max.words=50, random.order=FALSE, rot.per=0.15,
          colors=brewer.pal(8, "RdYlGn"))

Ồ, thật bất ngờ! machinelearning và datascience là những từ được nhắc đến nhiều nhất trong tất cả tweet của chúng ta—vốn dĩ đó chính là hai hashtag mà ta đã truyền. Vậy nên kết quả này là dễ đoán. Còn những từ khác thì thú vị hơn. Ví dụ, bigdata và artificialintelligence không nằm trong các hashtag khóa, nhưng lại xuất hiện với tần suất cao, nên có thể nói chúng thường được nhắc cùng với hai từ kia. Có những từ đáng chú ý khác mà ta thu được như python hay tensorflow, mang lại bối cảnh nhiều hơn về nội dung các tweet mà ta đã thu thập, vượt ra ngoài phạm vi hashtag.

Giờ hãy chuyển sang một phân tích đơn giản khác. Trong khoảng truyền 12 giờ, vào thời điểm nào ta thu thập được nhiều tweet nhất? Để làm vậy, ta sẽ lấy các mốc thời gian dạng số nguyên, chuyển chúng thành định dạng phù hợp, rồi vẽ số lượng tweet theo thời gian:

# Select the dates in which the tweets were created and convert them into UTC date-time format
all_tweets <- dbGetQuery(conn, "SELECT Tweet_ID, Date_Created FROM Tweet_Data")
all_tweets$Date_Created <- as.POSIXct(all_tweets$Date_Created, origin = "1970-01-01", tz = "UTC")

# Group by the day and hour and count the number of tweets that occurred in each bucket
all_tweets_2 <- all_tweets %>%
    mutate(day = day(Date_Created),
           month = month(Date_Created, label = TRUE),
           hour = hour(Date_Created)) %>%
    mutate(day_hour = paste(month,"-",day,"-",hour, sep = "")) %>%
    group_by(day_hour) %>%
    tally()

# Simple line ggplot
ggplot(all_tweets_2, aes(x = day_hour, y = n)) +
  geom_line(aes(group = 1)) +
  theme_minimal() +
  ggtitle("Tweet Freqeuncy During the 12-h Streming Period")+
  ylab("Tweet Count")+
  xlab("Month-Day-Hour")

Tuyệt! Giờ ta có thể thấy lượng tweet độc nhất lớn nhất thu được là vào ngày 2 tháng 9, trong khoảng từ 8:00 PM UTC đến 8:59 PM UTC (hiển thị theo giờ 24h là 20).

Kết luận

Chúc mừng! Giờ bạn đã biết cách xây dựng một pipeline ETL đơn giản trong R. Hai phân tích chúng ta thực hiện là những phân tích rất cơ bản với dữ liệu Twitter. Tuy nhiên, như đã đề cập, có rất nhiều việc để làm miễn là bạn xây dựng được pipeline vững chắc để đưa dữ liệu vào. Đây là khía cạnh chính của hướng dẫn này.

Nói vậy, hướng dẫn này chỉ trình bày một nghiên cứu tình huống quy mô nhỏ để đi qua quy trình xây dựng pipeline ETL cho dữ liệu Twitter. Xây dựng các pipeline ETL vững chắc và có thể mở rộng cho cả doanh nghiệp là một nỗ lực phức tạp, đòi hỏi tài nguyên tính toán và kiến thức sâu rộng, đặc biệt khi liên quan đến big data.

Tôi khuyến khích bạn nghiên cứu thêm và thử xây dựng các pipeline quy mô nhỏ của riêng mình, có thể bao gồm cả việc xây dựng bằng Python. Thậm chí, hãy thử một vài dự án big data. Ví dụ, DataCamp đã có các khóa học như Big Data Fundamentals via PySpark, nơi họ trình bày big data bằng các công cụ như PySpark để bạn mở rộng kiến thức trong lĩnh vực này.

Nếu bạn muốn học về kỹ thuật dữ liệu, hãy tham gia khóa Introduction to Data Engineering của DataCamp. Và nếu bạn đã sẵn sàng chứng minh kỹ năng mới với nhà tuyển dụng, hãy xem Chứng chỉ Data Engineer của chúng tôi.

Tài liệu tham khảo