R का उपयोग करके डेटा पाइपलाइन बनाना

यह तर्क दिया जा सकता है कि उचित ETL पाइपलाइनें डेटा विज्ञान का एक महत्वपूर्ण अंग हैं। साफ-सुथरे और व्यवस्थित डेटा के बिना, ऐसे गुणवत्तापूर्ण इनसाइट्स निकालना कठिन हो जाता है जो व्यावसायिक निर्णयों को बेहतर बनाएं।

इसलिए, इस ट्यूटोरियल में, हम यह जानेंगे कि R का उपयोग करके रीयल-टाइम ट्वीट्स को सीधे एक SQLite डेटाबेस में स्ट्रीम करने के लिए एक सरल ETL पाइपलाइन बनाना क्या-क्या शामिल करता है। यह, उदाहरण के लिए, सोशल नेटवर्क विश्लेषण में अक्सर किया जाने वाला कार्य है।

ध्यान डेटा संग्रहण और भंडारण की सोच-प्रक्रिया पर होगा, साथ ही R में rtweet पैकेज का उपयोग करके Twitter API को कैसे संचालित करें, इस पर भी।

शुरू करने के लिए, आइए सभी सही टूल्स तैयार रखें। सबसे पहले, आपको Twitter API तक अपनी पहुँच सेटअप करनी होगी। कुल मिलाकर, आपको ये चरण अपनाने होंगे:

यदि आपका Twitter खाता नहीं है, तो एक बनाएं।
इस लिंक पर जाकर डेवलपर अकाउंट के लिए आवेदन करें (ध्यान दें कि अब यह प्रक्रिया एक ऐसे आवेदन की मांग करती है जिसे Twitter को स्वीकृत करना होता है)।
निम्न वेबपेज पर एक नया ऐप बनाएं।
अपने ऐप के सभी विवरण भरें और अपना ऐक्सेस टोकन बनाएँ।
अपना कंज्यूमर की, कंज्यूमर सीक्रेट, ऐक्सेस टोकन और ऐक्सेस टोकन सीक्रेट एकत्र करें, जिनकी आपको API से कनेक्ट करने के लिए आवश्यकता होगी।

एक बार जब आप Twitter API के साथ तैयार हो जाएं, तो यदि आपके पास SQLite नहीं है तो आपको उसे प्राप्त करना होगा। अपने कंप्यूटर पर SQLite कैसे प्राप्त करें, इसकी पूरी प्रक्रिया के लिए DataCamp पर Beginners Guide to SQLite ट्यूटोरियल का पालन करें। इस ट्यूटोरियल के लिए SQLite चुनने का कारण इसका सरल संचालन है।

चरण 1: Twitter डेटा संग्रहीत करने के लिए एक डेटाबेस और तालिका बनाना

Twitter API ऐक्सेस और SQLite दोनों इंस्टॉल होने के साथ, हम अंततः एक पाइपलाइन बनाना शुरू कर सकते हैं जो समय के साथ स्ट्रीम होते ट्वीट्स को संग्रहीत करे। शुरू करने के लिए, हम नीचे बताए अनुसार R का उपयोग करके अपना नया SQLite डेटाबेस बनाएंगे:

# 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")
संचालित द्वारा

इसके बाद, हम डेटाबेस के अंदर ट्वीट्स रखने के लिए एक तालिका बना सकते हैं। हमारे विशेष मामले में, हम निम्न वेरिएबल्स संग्रहीत करेंगे:

Tweet_ID एक INTEGER प्राथमिक कुंजी के रूप में
User TEXT के रूप में
Tweet_Content TEXT के रूप में
Date_Created INTEGER के रूप में

आप सोच सकते हैं, मैं तारीखों को इंटीजर के रूप में क्यों सेट करता हूँ? इसका कारण यह है कि SQLite में तारीखों और समय के लिए आरक्षित डेटा प्रकार नहीं है। इसलिए, तारीखें 1970-01-01 से बीते सेकंड की संख्या के रूप में संग्रहीत होंगी।

अब आइए तालिका लिखते हैं:

dbExecute(conn, "CREATE TABLE Tweet_Data(
                  Tweet_ID INTEGER PRIMARY KEY,
                  User TEXT,
                  Tweet_Content TEXT,
                  Date_Created INTEGER)")
संचालित द्वारा

तालिका बनाने के बाद, आप sqlite3.exe पर जाकर जांच सकते हैं कि यह वास्तव में बन गई है। इसका स्क्रीनशॉट आप नीचे देख सकते हैं:

चरण 2: अपने पसंदीदा विषयों पर ट्वीट्स स्ट्रीम करें!

मानें या न मानें, एक सरल, कार्यात्मक Twitter स्ट्रीमिंग पाइपलाइन के लिए आवश्यकताओं और इन्फ्रास्ट्रक्चर के लिहाज से आप तैयार हैं। अब हमें API का उपयोग करके ट्वीट्स स्ट्रीम करने हैं। ध्यान देने योग्य है कि इस ट्यूटोरियल के उद्देश्यों के लिए, मैं मानक नि:शुल्क API का उपयोग करूँगा। प्रीमियम, सशुल्क वर्ज़न भी हैं जो, उदाहरण के लिए, यदि आप शोध करते हैं, तो आपकी ट्वीट स्ट्रीमिंग आवश्यकताओं के लिए बेहतर हो सकते हैं।

बिना देर किए, आइए अपना Twitter लिस्नर सेटअप करने की प्रक्रिया शुरू करते हैं। सबसे पहले, आपको rtweet पैकेज इम्पोर्ट करना होगा और अपने एप्लिकेशन के ऐक्सेस टोकन और सीक्रेट्स इनपुट करने होंगे, जैसा कि शुरुआत में बताया गया था:

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')
संचालित द्वारा

टोकन सेट होने के बाद, अगला कदम यह तय करना है कि आप कौन से ट्वीट्स स्ट्रीम करना चाहते हैं (यानी, किसे सुनना चाहते हैं)। rtweet पैकेज का stream_tweets फ़ंक्शन हमें Twitter API से क्वेरी करने के लिए कई विकल्प देता है। उदाहरण के लिए, आप दिए गए हैशटैग्स या कीवर्ड्स (400 तक) में से एक या अधिक वाले ट्वीट्स स्ट्रीम कर सकते हैं, सभी सार्वजनिक रूप से उपलब्ध ट्वीट्स का एक छोटा रैंडम उपसमुच्चय, यूज़र IDs या स्क्रीन (यूज़र) नामों (5000 तक) के समूह के ट्वीट्स को ट्रैक कर सकते हैं, या भौगोलिक लोकेशन के आधार पर ट्वीट्स एकत्र कर सकते हैं।

इस ट्यूटोरियल के लिए, मैंने डेटा साइंस से संबंधित हैशटैग्स वाले ट्वीट्स स्ट्रीम करने का निर्णय लिया (नीचे सूची देखें)। आप देखेंगे कि जिन हैशटैग्स को स्ट्रीम करने के लिए मैंने परिभाषित किया है, उनका फ़ॉर्मेट थोड़ा अलग है। हालाँकि, जब आप हैशटैग्स या कीवर्ड्स को सुनना चाहते हैं, तो stream_tweets फ़ंक्शन द्वारा इसी फ़ॉर्मेट की आवश्यकता होती है। यदि आप किसी निश्चित यूज़र सेट या कोऑर्डिनेट्स के आधार पर सुनना चाहते हैं, तो यह फ़ॉर्मेट बदलता है। अधिक विवरण के लिए डॉक्यूमेंटेशन देखें।

keys <- "#nlp,#machinelearning,#datascience,#chatbots,#naturallanguageprocessing,#deeplearning"
संचालित द्वारा

कीवर्ड्स तय होने के साथ, अब ट्वीट स्ट्रीमिंग लूप परिभाषित करने का समय है। इसे करने के कई तरीके हैं, लेकिन यह फ़ॉर्मेट अतीत में मेरे लिए अच्छा काम करता रहा है:

# 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
}
संचालित द्वारा

मूल रूप से, यह लूप 2-2 घंटे के अंतराल में कुल 12 घंटे तक, की-स्ट्रिंग्स में दिए गए किसी भी हैशटैग का उल्लेख करने वाले जितने संभव हो उतने ट्वीट्स स्ट्रीम करता है। हर 2 घंटे में, Twitter लिस्नर आपके वर्तमान वर्किंग डायरेक्टरी में filename वेरिएबल में निर्दिष्ट नाम के साथ एक .json फ़ाइल बनाता है।

फिर, यह इस फ़ाइलनाम को transform_and_clean_tweets फ़ंक्शन को पास करता है, जो इच्छानुसार रिट्वीट्स हटाता है, Twitter API द्वारा दी गई सभी कॉलमों में से वे चुनता है जिन्हें हम रखना चाहते हैं, और ट्वीट्स में निहित टेक्स्ट को सामान्यीकृत करता है।

इसके बाद, यह परिणामस्वरूप बने डेटा फ्रेम को हमारे SQLite डेटाबेस के अंदर पहले बनाई गई Tweet_Data तालिका में जोड़ देता है। अंत में, यह घंटों के काउंटर में 2 जोड़ता है (क्योंकि स्ट्रीम 2 घंटे तक चलती है), और बनाई गई .json फ़ाइल को हटा देता है। यह इसलिए किया जाता है क्योंकि अब हमारी सभी रुचि का डेटा हमारे डेटाबेस में है, और .json फ़ाइलें रखना स्टोरेज पर बोझ बन सकता है।

अब आइए 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)
}
संचालित द्वारा

जैसा कि नोट किया गया, यह फ़ंक्शन इच्छानुसार रिट्वीट्स को फ़िल्टर करने, वांछित फीचर्स बनाए रखने और ट्वीट टेक्स्ट को सामान्यीकृत करने के लिए बनाया गया है। मूल रूप से, इसे ETL संक्षेप के "T" भाग (ट्रांसफॉर्मेशन) के रूप में माना जा सकता है। इस फ़ंक्शन का एक प्रमुख घटक ट्वीट टेक्स्ट को साफ़ करने की प्रक्रिया है।

आम तौर पर, किसी भी प्रकार के विश्लेषण के लिए तैयार होने से पहले टेक्स्ट डेटा को कुछ प्री-प्रोसेसिंग चरणों की आवश्यकता होती है। ट्वीट्स के मामले में, इन चरणों में URLs, स्टॉप वर्ड्स और मेंशन्स को हटाना, टेक्स्ट को लोअर केस में बदलना, शब्दों का स्टेमिंग करना आदि शामिल हो सकते हैं। हालाँकि, ये सभी चरण हर समय आवश्यक नहीं होते। फिर भी, फिलहाल, मैं नीचे वे ट्वीट्स प्री-प्रोसेस करने के लिए उपयोग किए गए normalize_text फ़ंक्शन को प्रदर्शित करूंगा:

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)
}
संचालित द्वारा

आपके उपयोग-केस के आधार पर, ये चरण पर्याप्त हो सकते हैं। जैसा कि पहले कहा गया, आप स्टेमिंग या लैमेटाइज़ेशन जैसे और/या भिन्न चरण जोड़ना चुन सकते हैं, या केवल अक्षरों को रखना (अक्षरों और संख्याओं के बजाय) पसंद कर सकते हैं। मैं आपको प्रयोग करने और अलग-अलग संयोजनों को आज़माने के लिए प्रोत्साहित करता हूँ। यह regex कौशल का अभ्यास करने के लिए एक बेहतरीन जगह हो सकती है।

इन सभी चरणों के बाद, परिणामी स्थिति एक ऐसा SQLite डेटाबेस है जो सभी स्ट्रीम किए गए ट्वीट्स से भरा हुआ है। आप, उदाहरण के लिए, कुछ सरल क्वेरी चलाकर सत्यापित कर सकते हैं कि सब कुछ सही ढंग से काम किया:

data_test <- dbGetQuery(conn, "SELECT * FROM Tweet_Data LIMIT 20")
unique_rows <- dbGetQuery(conn, "SELECT COUNT() AS Total FROM Tweet_Data")
kable(data_test)
संचालित द्वारा

सभी स्ट्रीम किए गए ट्वीट्स से भरा SQLite डेटाबेस

print(as.numeric(unique_rows))
संचालित द्वारा

## [1] 1863
संचालित द्वारा

चरण 3: विश्लेषण

जब यह सुनिश्चित हो जाए कि ETL प्रक्रिया इच्छानुसार काम कर रही है, तो अंतिम चरण है कुछ इनसाइट्स निकालना और आपके द्वारा एकत्रित डेटा का विश्लेषण करना। उदाहरण के लिए, हमने जो ट्वीट्स एकत्र किए हैं, उनके लिए हम कुछ अलग-अलग चीज़ें आज़मा सकते हैं; ट्वीट की सामग्री में उल्लिखित शब्दों का वर्डक्लाउड, और एक टाइमलाइन जो यह दिखाए कि हमारे 12-घंटे के स्ट्रीमिंग अवधि में किस समय हमें सबसे अधिक ट्वीट्स प्राप्त हुए। यह सूची स्पष्ट रूप से उन सभी प्रकार के शोधों का व्यापक अवलोकन नहीं है जो ट्वीट्स का उपयोग करके किए जा सकते हैं, जो सेंटिमेंट एनालिसिस से लेकर साइकोग्राफिक स्टडीज़ और आगे तक जा सकते हैं।

यह कहते हुए, चलिए सीधे एक अच्छा वर्डक्लाउड बनाने पर चलते हैं:

# 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"))
संचालित द्वारा

वाह, क्या भारी आश्चर्य! machinelearning और datascience हमारे सभी ट्वीट्स में सबसे अधिक उल्लिखित शब्द हैं—ये वही दो हैशटैग हैं जिनके लिए हम स्ट्रीम कर रहे थे। अतः, यह परिणाम अपेक्षित था। बाकी शब्द देखने में अधिक रोचक हैं। उदाहरण के लिए, bigdata और artificialintelligence हमारे कीज़ में हैशटैग नहीं थे, फिर भी वे उच्च आवृत्ति के साथ प्रकट हुए, इसलिए यह तर्क दिया जा सकता है कि इन पर भी अन्य दो के साथ अक्सर चर्चा होती है। हमने अन्य दिलचस्प शब्द भी उठाए जैसे python या tensorflow जो हमें सिर्फ हैशटैग्स से परे, हमारे द्वारा एकत्र किए गए ट्वीट्स की सामग्री के बारे में थोड़ा और संदर्भ देते हैं।

अब एक और सरल विश्लेषण की ओर मुड़ते हैं। हमारी 12-घंटे की स्ट्रीम में, हमने सबसे अधिक ट्वीट्स किस समय एकत्र किए? इसके लिए, हम अपनी इंटीजर तिथियाँ इकट्ठा करेंगे, उन्हें उचित फ़ॉर्मेट में बदलेंगे, और फिर समय के साथ ट्वीट्स की मात्रा का प्लॉट बनाएंगे:

# 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")
संचालित द्वारा

बहुत बढ़िया! अब हम देख सकते हैं कि हमें सबसे अधिक अद्वितीय ट्वीट्स 2 सितंबर को 8:00 PM UTC से 8:59 PM UTC के बीच मिले (यह सैन्य समय में 20 के रूप में दिखता है)।

निष्कर्ष

बधाई हो! अब आप जानते हैं कि R में एक सरल ETL पाइपलाइन कैसे बनाते हैं। हमने जो दो विश्लेषण किए, वे Twitter डेटा का उपयोग करके किए जाने वाले बहुत बुनियादी विश्लेषणों का प्रतिनिधित्व करते हैं। हालाँकि, जैसा कि पहले उल्लेख किया गया, जैसे ही आप डेटा लाने के लिए एक मज़बूत पाइपलाइन बनाते हैं, करने के लिए बहुत कुछ है। यही पहलू इस ट्यूटोरियल का मुख्य फोकस था।

यह कहते हुए, यह ट्यूटोरियल Twitter डेटा के लिए ETL पाइपलाइन बनाने की प्रक्रिया पर चलने हेतु केवल एक बहुत छोटे पैमाने का केस स्टडी दिखाता है। पूरे एंटरप्राइज़ के लिए मज़बूत और स्केलेबल ETL पाइपलाइन बनाना एक जटिल कार्य है, जिसके लिए व्यापक कम्प्यूटिंग संसाधनों और ज्ञान की आवश्यकता होती है, खासकर जब बिग डेटा शामिल हो।

मैं आपको आगे शोध करने और अपनी स्वयं की छोटे पैमाने की पाइपलाइनें बनाने की कोशिश करने के लिए प्रोत्साहित करता हूँ, जिसमें Python में एक बनाना भी शामिल हो सकता है। शायद, आगे बढ़ें और कुछ बिग डेटा प्रोजेक्ट्स भी आज़माएँ। उदाहरण के लिए, DataCamp में पहले से ऐसे कोर्स हैं जैसे यह Big Data Fundamentals via PySpark कोर्स, जहाँ वे PySpark जैसे टूल्स का उपयोग करके बिग डेटा पर जाते हैं, जो इस क्षेत्र में आपके ज्ञान को और आगे बढ़ा सकते हैं।

यदि आप डेटा इंजीनियरिंग के बारे में सीखना चाहते हैं, तो DataCamp का Introduction to Data Engineering कोर्स लें। और यदि आप नियोक्ताओं के सामने अपने नए कौशल सिद्ध करने के लिए तैयार हैं, तो हमारी Data Engineer Certification देखें।

संदर्भ

Foley, D. (2019, May 11). Streaming Twitter Data into a MySQL Database. प्राप्त किया गया: https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6

विषय

R डेटा साइंस डेटा इंजीनियरिंग मशीन लर्निंग SQL

प्रौद्योगिकियाँ

विषय

R का उपयोग करके डेटा पाइपलाइन बनाना

AI के साथ खोजें

चरण 1: Twitter डेटा संग्रहीत करने के लिए एक डेटाबेस और तालिका बनाना

चरण 2: अपने पसंदीदा विषयों पर ट्वीट्स स्ट्रीम करें!

चरण 3: विश्लेषण

निष्कर्ष

विषय

R के बारे में और जानें

PySpark के साथ Big Data Fundamentals

Introduction to Data Engineering

R परिचय

प्रौद्योगिकियाँ

विषय

R के बारे में और जानें

PySpark के साथ Big Data Fundamentals

Introduction to Data Engineering

R परिचय