Membangun Data Pipeline menggunakan R

Seseorang bisa berpendapat bahwa pipeline ETL yang tepat adalah organ vital dari ilmu data. Tanpa data yang bersih dan terorganisir, akan sulit menghasilkan insight berkualitas yang meningkatkan pengambilan keputusan bisnis.

Oleh karena itu, dalam tutorial ini, kita akan mengeksplorasi apa saja yang diperlukan untuk membangun pipeline ETL sederhana guna melakukan streaming Tweet secara real-time langsung ke database SQLite menggunakan R. Ini adalah tugas yang cukup umum dalam analisis jejaring sosial, misalnya.

Fokusnya adalah membahas proses berpikir dalam pengumpulan dan penyimpanan data serta cara mengoperasikan Twitter API menggunakan paket rtweet di R.

Untuk memulai, mari pastikan semua peralatan yang tepat tersedia. Hal pertama yang Anda perlukan adalah menyiapkan akses ke Twitter API. Secara umum, Anda perlu mengikuti langkah-langkah berikut:

• Buat akun Twitter, jika Anda belum punya.
• Ikuti tautan ini dan ajukan permohonan akun pengembang (perhatikan bahwa proses ini kini memerlukan aplikasi yang harus disetujui oleh Twitter).
• Buat aplikasi baru pada laman web berikut.
• Isi semua detail tentang aplikasi Anda dan buat access token.
• Kumpulkan consumer key, consumer secret, access token, dan access token secret yang Anda perlukan untuk terhubung ke API.

Setelah Anda siap dengan Twitter API, Anda perlu mendapatkan SQLite jika belum memilikinya. Untuk proses lengkap cara memasang SQLite di komputer Anda, ikuti tutorial Beginners Guide to SQLite di DataCamp. Pilihan menggunakan SQLite untuk tutorial ini dibuat karena kemudahannya dalam pengoperasian.

Langkah 1: Membuat Database dan Tabel untuk Menyimpan Data Twitter

Dengan akses Twitter API dan SQLite telah terpasang, kita akhirnya bisa mulai membangun pipeline untuk menyimpan Tweet saat kita melakukan streaming dari waktu ke waktu. Untuk memulai, kita akan membuat database SQLite baru menggunakan R seperti yang diuraikan di bawah ini:

# 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")

Setelah itu, kita dapat membuat tabel di dalam database untuk menampung tweet. Dalam kasus kita, kita akan menyimpan variabel-variabel berikut:

• Tweet_ID sebagai INTEGER primary key
• User sebagai TEXT
• Tweet_Content sebagai TEXT
• Date_Created sebagai INTEGER

Anda mungkin bertanya-tanya, mengapa tanggal saya set sebagai integer? Ini karena SQLite tidak memiliki tipe data khusus untuk tanggal dan waktu. Oleh karena itu, tanggal akan disimpan sebagai jumlah detik sejak 1970-01-01.

Sekarang mari kita lanjut untuk menulis tabelnya:

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

Setelah Anda membuat tabel, Anda dapat membuka sqlite3.exe dan memeriksa bahwa tabel tersebut memang telah dibuat. Anda dapat melihat tangkapan layarnya di bawah ini:

Langkah 2: Streaming Tweet tentang Topik Favorit Anda!

Percaya atau tidak, persyaratan dan infrastruktur yang dibutuhkan untuk memiliki pipeline streaming Twitter yang sederhana dan fungsional sudah selesai. Yang perlu kita lakukan sekarang adalah melakukan streaming Tweet menggunakan API. Perlu dicatat bahwa untuk tujuan tutorial ini, saya akan menggunakan API standar gratis. Ada versi berbayar premium yang mungkin lebih sesuai dengan kebutuhan streaming Tweet Anda jika Anda melakukan riset, misalnya.

Tanpa berlama-lama, mari kita mulai proses menyiapkan pendengar Twitter kita. Hal pertama yang Anda perlukan adalah mengimpor paket rtweet dan memasukkan access token dan secret aplikasi Anda seperti dijelaskan di awal:

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')

Dengan token Anda siap, selanjutnya adalah memutuskan tweet seperti apa yang ingin Anda stream (yaitu, didengarkan). Fungsi stream_tweets dalam paket rtweet memberi kita berbagai opsi untuk melakukan kueri ke Twitter API. Misalnya, Anda dapat melakukan streaming tweet yang berisi satu atau lebih dari sekumpulan hashtag atau kata kunci tertentu (hingga 400), subset acak kecil dari semua tweet publik yang tersedia, melacak tweet dari sekelompok user ID atau screen name (nama pengguna) (hingga 5000), atau mengumpulkan tweet berdasarkan lokasi geografis.

Untuk tutorial ini, saya memutuskan untuk langsung melakukan streaming tweet yang mengandung hashtag terkait data science (lihat daftar di bawah). Anda mungkin memperhatikan bahwa format saya mendefinisikan hashtag untuk di-stream agak aneh. Namun, itulah format yang diperlukan oleh fungsi stream_tweets saat Anda ingin mendengarkan hashtag atau kata kunci. Format ini berbeda jika Anda bermaksud mendengarkan sekumpulan pengguna tertentu atau berdasarkan koordinat. Untuk detailnya, lihat dokumentasi.

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

Dengan kata kunci yang sudah ditentukan, saatnya mendefinisikan loop streaming tweet. Ada beberapa cara untuk melakukannya, tetapi format ini bekerja dengan baik bagi saya sebelumnya:

# 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
}

Pada dasarnya, loop tersebut melakukan streaming sebanyak mungkin tweet yang menyebut salah satu hashtag dalam string key dalam interval 2 jam selama total 12 jam. Setiap 2 jam, pendengar Twitter membuat file .json di direktori kerja Anda saat ini dengan nama yang ditetapkan dalam variabel filename.

Kemudian, nama file ini diteruskan ke fungsi transform_and_clean_tweets yang menghapus retweet jika diinginkan, memilih kolom yang ingin kita simpan dari semua yang diberikan oleh Twitter API, dan menormalkan teks yang terdapat dalam Tweet.

Kemudian, ia menambahkan dataframe hasilnya ke tabel Tweet_Data yang kita buat sebelumnya di dalam database SQLite kita. Terakhir, ia menambahkan 2 ke penghitung jam (karena setiap stream berlangsung selama 2 jam), dan menghapus file .json yang dibuat. Ini dilakukan karena semua data yang diinginkan kini berada di database kita, dan menyimpan file .json dapat menjadi beban penyimpanan.

Sekarang mari melihat lebih dekat fungsi transform_and_clean_tweets secara rinci:

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)
}

Seperti disebutkan, fungsi ini dimaksudkan untuk memfilter retweet jika diinginkan, mempertahankan fitur yang diinginkan, dan menormalkan teks tweet. Intinya, ini dapat dianggap sebagai bagian "T" dari akronim ETL (transformasi). Satu komponen kunci dari fungsi ini adalah proses pembersihan teks tweet.

Biasanya, data teks memerlukan beberapa langkah prapemrosesan sebelum siap untuk dianalisis. Dalam kasus tweet, langkah-langkah ini dapat mencakup penghapusan URL, stop word dan mention, mengubah teks menjadi huruf kecil, stemming kata, dan sebagainya. Namun, tidak semua langkah ini selalu diperlukan. Meski begitu, untuk saat ini, saya akan menampilkan fungsi normalize_text yang saya gunakan untuk memproses tweet berikut ini:

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)
}

Tergantung pada kasus penggunaan Anda, langkah-langkah ini mungkin sudah cukup. Seperti disebutkan sebelumnya, Anda bisa menambahkan langkah lain seperti stemming atau lemmatization atau hanya mempertahankan huruf alih-alih huruf dan angka. Saya mendorong Anda untuk bereksperimen dan mencoba berbagai kombinasi. Ini bisa menjadi tempat yang bagus untuk melatih keterampilan regex.

Setelah semua langkah ini, kondisi akhirnya adalah database SQLite yang terisi dengan semua tweet yang di-stream. Anda dapat memvalidasi bahwa semuanya berjalan dengan benar dengan menjalankan beberapa kueri sederhana, misalnya:

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

print(as.numeric(unique_rows))

## [1] 1863

Langkah 3: Analisis

Ketika sudah yakin bahwa proses ETL berjalan sebagaimana mestinya, langkah terakhir adalah menggali beberapa insight dan menganalisis data yang telah Anda kumpulkan. Misalnya, untuk tweet yang telah kita kumpulkan, kita bisa mencoba beberapa hal berbeda: wordcloud dari istilah yang disebutkan dalam konten tweet, dan timeline untuk memvisualisasikan kapan dalam periode 12 jam streaming kita memperoleh jumlah tweet terbanyak. Daftar ini jelas bukan gambaran menyeluruh dari seluruh penelitian yang dapat dilakukan menggunakan tweet, yang dapat berkisar dari analisis sentimen hingga studi psikografis dan seterusnya.

Dengan demikian, mari langsung membuat wordcloud yang menarik:

# 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"))

Wah, sungguh kejutan besar! machinelearning dan datascience adalah kata yang paling sering disebutkan di semua tweet kita—kebetulan itu adalah dua hashtag yang kita streaming. Jadi, hasil ini memang sudah diperkirakan. Kata-kata lainnya lebih menarik untuk dilihat. Misalnya, bigdata dan artificialintelligence bukan hashtag dalam keys kita, namun muncul dengan frekuensi tinggi, sehingga bisa dikatakan bahwa keduanya sering dibicarakan bersama dengan dua yang lain. Ada kata-kata menarik lainnya yang kita tangkap seperti python atau tensorflow yang memberi kita sedikit lebih banyak konteks tentang isi tweet yang kita kumpulkan, tidak hanya sekadar hashtag.

Sekarang mari beralih ke analisis sederhana lainnya. Dalam stream 12 jam kita, pada jam berapa kita mengumpulkan tweet paling banyak? Untuk itu, kita akan mengumpulkan tanggal dalam bentuk integer, mengubahnya ke format yang benar, lalu memplot jumlah tweet dari waktu ke waktu:

# 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")

Mantap! Kini kita dapat melihat bahwa jumlah tweet unik terbanyak yang kita peroleh berasal dari tanggal 2 September antara pukul 20.00 UTC hingga 20.59 UTC (muncul dalam format waktu 24 jam sebagai 20).

Kesimpulan

Selamat! Sekarang Anda tahu cara membangun pipeline ETL sederhana di R. Dua analisis yang kita lakukan mewakili analisis yang sangat dasar menggunakan data Twitter. Namun, seperti disebutkan sebelumnya, ada banyak hal yang dapat dilakukan selama Anda membangun pipeline yang andal untuk memasukkan data. Aspek ini menjadi fokus utama tutorial ini.

Dengan demikian, tutorial ini hanya menunjukkan studi kasus skala sangat kecil untuk menelusuri proses membangun pipeline ETL untuk data Twitter. Membangun pipeline ETL yang andal dan dapat diskalakan untuk seluruh perusahaan adalah upaya yang rumit yang memerlukan sumber daya komputasi dan pengetahuan yang luas, terutama ketika melibatkan big data.

Saya mendorong Anda untuk melakukan riset lebih lanjut dan mencoba membangun pipeline skala kecil Anda sendiri, yang bisa mencakup membangunnya di Python. Bahkan, coba langsung beberapa proyek big data. Misalnya, DataCamp sudah memiliki beberapa kursus seperti kursus Big Data Fundamentals via PySpark yang membahas big data menggunakan alat seperti PySpark untuk memperdalam pengetahuan Anda di bidang ini.

Jika Anda ingin mempelajari tentang rekayasa data, ikuti kursus Introduction to Data Engineering dari DataCamp. Dan jika Anda siap membuktikan keterampilan baru Anda kepada pemberi kerja, lihat Sertifikasi Data Engineer kami. 

Referensi

1. Foley, D. (2019, 11 Mei). Streaming Twitter Data into a MySQL Database. Diambil dari https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6