R kullanarak Veri Hatları (Data Pipeline) Oluşturma

Doğru ETL hatlarının veri biliminin hayati bir organı olduğu söylenebilir. Temiz ve düzenli veri olmadan, iş kararlarını geliştiren nitelikli içgörüler üretmek zorlaşır.

Bu nedenle, bu eğitimde R kullanarak gerçek zamanlı Tweet’leri doğrudan bir SQLite veritabanına aktaran basit bir ETL hattı oluşturmanın ne gerektirdiğini inceleyeceğiz. Bu, örneğin sosyal ağ analizinde oldukça yaygın bir görevdir.

Odak noktası, veri toplama ve depolama süreçlerinin düşünce yapısını kapsamanın yanı sıra R’de rtweet paketiyle Twitter API’sinin nasıl kullanılacağı üzerinde olacaktır.

Başlamak için, önce tüm doğru araçlara sahip olalım. İhtiyacınız olan ilk şey, Twitter API erişiminizi ayarlamaktır. Genel olarak şu adımları izlemeniz gerekir:

• Bir Twitter hesabı oluşturun, eğer yoksa.
• Şu bağlantıyı takip edin ve geliştirici hesabı için başvurun (bu sürecin artık Twitter’ın onaylaması gereken bir başvuru gerektirdiğini unutmayın).
• Aşağıdaki web sayfasında yeni bir uygulama oluşturun.
• Uygulamanızla ilgili tüm ayrıntıları doldurun ve erişim belirtecinizi (token) oluşturun.
• API’ye bağlanmak için ihtiyaç duyacağınız consumer key, consumer secret, access token ve access token secret bilgilerini edinin.

Twitter API ile işiniz bittiğinde ve eğer yoksa SQLite’ı edinmeniz gerekir. Bilgisayarınıza SQLite kurulumunun tam süreci için DataCamp’teki SQLite’a Yeni Başlayanlar Rehberi eğitimini takip edin. Bu eğitimde SQLite’ın tercih edilme nedeni, kullanımının basit olmasıdır.

Adım 1: Twitter Verilerini Saklamak için Veritabanı ve Tablo Oluşturma

Hem Twitter API erişimi hem de SQLite kurulu olduğuna göre, artık zaman içinde akıtırken Tweet’leri depolayacak bir hat oluşturmaya başlayabiliriz. Başlamak için, aşağıda gösterildiği gibi R kullanarak yeni SQLite veritabanımızı oluşturacağız:

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

Ardından, veritabanı içinde tweet’leri tutacak bir tablo oluşturabiliriz. Bizim özel durumumuzda aşağıdaki değişkenleri saklayacağız:

• Tweet_ID birincil anahtar olan INTEGER
• User TEXT
• Tweet_Content TEXT
• Date_Created INTEGER

Şunu merak ediyor olabilirsiniz: Neden tarihleri tamsayı olarak ayarlıyorum? Bunun nedeni, SQLite’ın tarih ve saatler için ayrılmış bir veri türüne sahip olmamasıdır. Bu nedenle tarihler, 1970-01-01’den bu yana geçen saniye sayısı olarak saklanacaktır.

Şimdi tabloyu yazmaya geçelim:

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

Tabloyu oluşturduktan sonra, sqlite3.exe’ye gidip gerçekten oluşturulduğunu kontrol edebilirsiniz. Aşağıda bunun bir ekran görüntüsünü görebilirsiniz:

Adım 2: Favori Konularınız Hakkında Tweet’leri Akıtın!

İnanın ya da inanmayın, basit ve işlevsel bir Twitter akış hattı için gereken gereksinimler ve altyapı açısından işiniz bitti. Şimdi yapmamız gereken, API’yi kullanarak Tweet akıtmak. Bu eğitim kapsamında standart ücretsiz API’yi kullanacağımı belirtmekte fayda var. Örneğin araştırma yapıyorsanız, ihtiyaçlarınıza daha uygun olabilecek ücretli premium sürümler de mevcut.

Daha fazla gecikmeden, Twitter dinleyicimizi kurma sürecine başlayalım. Öncelikle rtweet paketini içe aktarmanız ve başta açıklandığı gibi uygulamanızın erişim belirteçlerini ve sırlarını girmeniz gerekir:

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

Belirteciniz hazır olduğuna göre, bir sonraki adım hangi tweet’leri akıtmak istediğinize (yani dinlemek istediğinize) karar vermektir. rtweet paketindeki stream_tweets işlevi, Twitter API’sini sorgulamak için çeşitli seçenekler sunar. Örneğin, verilen bir dizi hashtag veya anahtar kelimeyi (400’e kadar) içeren tweet’leri akıtabilir, herkese açık tüm tweet’lerin küçük rastgele bir alt kümesini alabilir, bir grup kullanıcı kimliğinin veya ekran (kullanıcı) adlarının (5000’e kadar) tweet’lerini izleyebilir ya da coğrafi konuma göre tweet toplayabilirsiniz.

Bu eğitim için, veri bilimiyle ilgili hashtag’leri içeren tweet’leri akıtmaya karar verdim (aşağıdaki listeye bakın). Akıtılacak hashtag’leri tanımladığım biçimin biraz tuhaf göründüğünü fark edebilirsiniz. Ancak, hashtag’leri veya anahtar kelimeleri dinlemek istediğinizde stream_tweets işlevinin gerektirdiği biçim budur. Belirli bir kullanıcı kümesini veya koordinatları dinlemek istiyorsanız bu biçim değişir. Daha fazla ayrıntı için belgelere bakın.

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

Anahtar kelimeler tanımlandığına göre, tweet akış döngüsünü tanımlama zamanı. Bunu yapmanın birkaç yolu var, ancak bu biçim geçmişte benim için iyi çalıştı:

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

Özetle, bu döngü toplam 12 saat boyunca 2 saatlik aralıklarla, anahtar dizgilerindeki herhangi bir hashtag’i anan olabildiğince çok tweet’i akıtır. Her 2 saatte bir, Twitter dinleyicisi geçerli çalışma dizininizde filename değişkeninde belirtilen adla bir .json dosyası oluşturur.

Daha sonra bu dosya adını, istenirse retweet’leri kaldıran, Twitter API’sinin sunduğu sütunlardan tutmak istediklerimizi seçen ve Tweet’lerde yer alan metni normalize eden transform_and_clean_tweets işlevine geçirir.

Ardından, ortaya çıkan veri çerçevesini SQLite veritabanımız içinde daha önce oluşturduğumuz Tweet_Data tablosuna ekler. Son olarak, saat sayacına 2 ekler (akışlar 2 saat sürdüğünden) ve oluşturulan .json dosyasını siler. Bu, ilgi duyulan tüm veriler artık veritabanımızda olduğundan ve .json dosyalarını tutmanın depolama yüküne dönüşebileceğinden yapılır.

Şimdi transform_and_clean_tweets işlevine daha yakından bakalım:

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

Belirtildiği gibi, bu işlev istenirse retweet’leri filtrelemek, istenen özellikleri korumak ve tweet metnini normalize etmek için tasarlanmıştır. Özünde bu, ETL kısaltmasının “T” kısmı (dönüştürme) olarak kabul edilebilir. Bu işlevin önemli bir bileşeni, tweet metninin temizlenmesi sürecidir.

Genellikle, metin verileri herhangi bir analize hazır hale gelmeden önce birkaç ön işleme adımı gerektirir. Tweet’ler söz konusu olduğunda bu adımlar; URL’lerin, durak sözcüklerin ve bahsetmelerin kaldırılması, metnin küçük harfe dönüştürülmesi, kök (stemming) çıkarımı vb. içerebilir. Ancak bu adımların tümü her zaman gerekli değildir. Yine de şimdilik, bu tweet’leri ön işlemek için kullandığım normalize_text işlevini aşağıda göstereceğim:

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

Kullanım durumunuza bağlı olarak bu adımlar yeterli olabilir. Daha önce de belirtildiği gibi, kök çıkarımı veya gövdeleme (lemmatization) gibi daha fazla veya farklı adım eklemeyi ya da harf ve sayılar yerine yalnızca harfleri tutmayı seçebilirsiniz. Deney yapmanızı ve farklı kombinasyonları denemenizi öneririm. Burası regex becerilerini uygulamak için harika bir alan olabilir.

Tüm bu adımlardan sonra ortaya çıkan durum, akıtılan tüm tweet’lerle doldurulmuş bir SQLite veritabanıdır. Her şeyin doğru çalıştığını, örneğin birkaç basit sorgu çalıştırarak doğrulayabilirsiniz:

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

Adım 3: Analiz

ETL sürecinin hedeflendiği gibi çalıştığından emin olduğunuzda, son adım bazı içgörüler elde etmek ve topladığınız verileri analiz etmektir. Örneğin, topladığımız tweet’ler için birkaç farklı şey deneyebiliriz; tweet içeriğinde geçen terimlerin bir kelime bulutu ve 12 saatlik akış dönemimizde en çok tweet’i ne zaman elde ettiğimizi görselleştirmek için bir zaman çizelgesi. Bu liste elbette tweet’ler kullanılarak yapılabilecek araştırmaların tüm yelpazesinin kapsamlı bir özeti değildir; duygu analizinden psikografik çalışmalara ve ötesine kadar uzanabilir.

Öyleyse, doğrudan hoş bir kelime bulutu oluşturmaya geçelim:

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

Vay canına, ne büyük sürpriz! machinelearning ve datascience, tüm tweet’lerimizde en çok geçen kelimeler—akışını yaptığımız iki hashtag olmaları tesadüf değil. Dolayısıyla bu sonuç bekleniyordu. Diğer kelimeler ise bakması biraz daha eğlenceli. Örneğin, bigdata ve artificialintelligence anahtarlarımızda hashtag değildi, yine de yüksek sıklıkla ortaya çıktılar; bu nedenle, diğer ikisiyle birlikte sıkça konuşuldukları söylenebilir. python veya tensorflow gibi topladığımız tweet’lerin yalnızca hashtag’lerin ötesindeki içeriğine biraz daha bağlam katan başka ilgi çekici kelimeler de yakaladık.

Şimdi başka bir basit analize geçelim. 12 saatlik akışımızda en çok tweet’i hangi saatte topladık? Bunun için tamsayı tarihleri toplayacak, uygun bir biçime dönüştürecek ve ardından zaman içindeki tweet miktarını çizeceğiz:

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

Harika! Artık 2 Eylül’de UTC ile 20:00 ve 20:59 (askeri saatte 20 olarak görünür) arasında en fazla sayıda benzersiz tweet’i elde ettiğimizi görebiliyoruz.

Sonuç

Tebrikler! Artık R’de basit bir ETL hattının nasıl oluşturulacağını biliyorsunuz. Yaptığımız iki analiz, Twitter verileri kullanılarak gerçekleştirilebilecek çok temel analizleri temsil ediyor. Ancak daha önce belirtildiği gibi, verileri içeri almak için sağlam bir hat kurduğunuz sürece yapılacak çok şey var. Bu yön, bu eğitimin ana odağıydı.

Bununla birlikte, bu eğitim Twitter verileri için ETL hatları oluşturma sürecini adım adım göstermek amacıyla yalnızca çok küçük ölçekli bir örnek olay sundu. Bir işletmenin tamamı için sağlam ve ölçeklenebilir ETL hatları kurmak, özellikle büyük veri söz konusu olduğunda, kapsamlı bilgi ve hesaplama kaynakları gerektiren karmaşık bir iştir.

Daha fazla araştırma yapmanızı ve kendi küçük ölçekli hatlarınızı kurmayı denemenizi teşvik ederim; buna Python’da bir hat kurmak da dahil olabilir. Hatta bazı büyük veri projelerini denemeye bile girişebilirsiniz. Örneğin DataCamp, PySpark gibi araçlar kullanarak büyük veriyi ele aldıkları PySpark ile Büyük Veri Temelleri gibi kurslara zaten sahip; bu kurslar bu alandaki bilginizi ilerletebilir.

Veri mühendisliği hakkında bilgi edinmek isterseniz, DataCamp’in Veri Mühendisliğine Giriş kursunu alın. Ve yeni becerilerinizi işverenlere kanıtlamaya hazırsanız, Veri Mühendisi Sertifikamıza göz atın. 

Kaynaklar

1. Foley, D. (11 Mayıs 2019). Streaming Twitter Data into a MySQL Database. Alındığı yer: https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6