Analýza sentimentu na Twitteri pomocou Spark

Akshay Nagpal2018-07-31 · 15 min read

Analýza sentimentu na Twitteri pomocou Spark

V našom predchádzajúcom príspevku, som vypracoval spôsob, ako extrahovať dáta z Twitteru v reálnom čase pomocou Apache Flume. V súčasnosti mám veľa dát z Twitteru. Preto by som ich chcel analyzovať a nájsť v nich nejaké trendy. Na vykonanie analýzy sentimentu dát z Twitteru použijem ďalší nástroj pre Big Data, Apache Spark.

Podľa Hortonworks, “Apache Spark je rýchly in-memory engine na spracovanie dát s elegantnými a expresívnymi vývojovými API, ktoré dátovým pracovníkom umožňujú efektívne spúšťať streamingové, strojovo-učiace alebo SQL úlohy vyžadujúce rýchly iteratívny prístup k dátovým sadám. Keď Spark beží na Apache Hadoop YARN, vývojári kdekoľvek na svete môžu vytvárať aplikácie, ktoré využívajú silu Sparku, získavajú poznatky a obohacujú svoje dátovo-vedecké úlohy v rámci jedinej zdieľanej dátovej sady v Hadoop.”

Program v Sparku môže byť napísaný v jazykoch JAVA, Scala, Python alebo R. V tomto prípade budeme používať JAVA spolu s Maven. Okrem toho sa Spark dodáva s distribúciou HDP aj Cloudera. Spark 2 je aktuálne používaná verzia.

Na vykonanie analýzy sentimentu pomocou Sparku vytváram nový Maven projekt. Pomenoval som ho „Twitter Sentiment Analyzer’“. Ďalej vytváram triedu „TwitterDataFlow.java“, v ktorej implementujem všetky požadované metódy.

Language Tool: Korektor pravopisu

Na začiatku, v rámci POC, som zistil, že ak je pravopis v tweetoch nesprávny, výsledky analýzy sentimentu sú nepriaznivo ovplyvnené. Preto zavádzam SpellChecker. Pomôže nám opraviť pravopis tweetov pred ich použitím na analýzu sentimentu.

Urobím to tak, že vytvorím novú statickú metódu s názvom ‘CorrectSpell’. Okrem toho použijem LanguageTool na kontrolu pravopisu a jeho opravu.

Podľa LanguageTool’s GIT, “LanguageTool je open-source softvér na korektúru pre angličtinu, francúzštinu, nemčinu, poľštinu, ruštinu a viac ako 20 ďalších jazykov. Nachádza mnoho chýb, ktoré jednoduchý korektor pravopisu nedokáže detegovať.”

Ďalej pridávam závislosť pre language tool do pom.xml:

<dependency>
<groupId>org.languagetool</groupId>
<artifactId>language-en</artifactId>
<version>4.0</version>
</dependency>

Potom definujem statickú premennú na úrovni triedy langTool triedy JLanguageTool. Následne inicializujem langTool objektom triedy AmericanEnglish.

Ďalej kódujem metódu s názvom SpellChecker so vstupom ako String text (bežný text) a návratovým typom taktiež ako String (text so správnym pravopisom). Používam metódu check triedy JLanguageTool s parametrom ako neskontrolovaný text. Následne vracia zoznam RuleMatch. Podľa Java Docs pre JLanguageTool trieda RuleMatch poskytuje “informácie o chybovom pravidle, ktoré sa zhoduje s textom, a pozíciu tejto zhody.”

Následne definujem tri premenné: ‘result’ typu String, ‘lastPos’ typu integer, ‘tmp’ typu String. Okrem toho pri každom RuleMatch znova vytváram vetu s prvým navrhovaným pravopisom z nástroja. Týmto som pridal potrebné bloky try-catch všade, kde to bolo potrebné.

package com.CloudSigma.Spark.TwitterSentiment;

import java.util.List;
import org.languagetool.JLanguageTool;
import org.languagetool.language.AmericanEnglish;
import org.languagetool.rules.RuleMatch;


public class LanguageCheck {
	
	static JLanguageTool langTool = new JLanguageTool(new AmericanEnglish());

    public static String CorrectSpell(String text) {
 	
    	 String query = text;
    	 
         try {
             List matches = langTool.check(query);
             String result = "";

             int lastPos = 0;
             String tmp = "";

             for (RuleMatch ma : matches) {

                 try {

                     tmp = ma.getSuggestedReplacements().get(0);
                     result += query.substring(lastPos, ma.getFromPos());
                     result += tmp;
                     lastPos = ma.getToPos();

                 } catch (Exception e) {
                     return text;
                 }
             }

             if (lastPos &lt; query.length()) {
                 result += query.substring(lastPos, query.length());
             }

             return result;

         } catch (Exception e) {
             return text;
         }
    	
    }
    
}

package com.CloudSigma.Spark.TwitterSentiment;

import java.util.List;

import org.languagetool.JLanguageTool;

import org.languagetool.language.AmericanEnglish;

import org.languagetool.rules.RuleMatch;

public class LanguageCheck {

static JLanguageTool langTool = new JLanguageTool(new AmericanEnglish());

public static String CorrectSpell(String text) {

String query = text;

try {

List matches = langTool.check(query);

String result = "";

int lastPos = 0;

String tmp = "";

for (RuleMatch ma : matches) {

try {

tmp = ma.getSuggestedReplacements().get(0);

result += query.substring(lastPos, ma.getFromPos());

result += tmp;

lastPos = ma.getToPos();

} catch (Exception e) {

return text;

}

if (lastPos < query.length()) {

result += query.substring(lastPos, query.length());

}

return result;

} catch (Exception e) {

return text;

}

Analyzátor sentimentu: Stanford CoreNLP

Ďalším krokom v analýze sentimentu pomocou Spark je nájsť sentimenty v texte. Aby som to dosiahol, používam knižnicu Stanford’s Core NLP na hľadanie hodnôt sentimentu. Potom vytvorím triedu s názvom ‘StanfordSentiment’, v ktorej budem implementovať túto knižnicu na hľadanie sentimentov v našom texte.

Na to pridám do súboru pom.xml nasledujúce závislosti:

<!– Toto je knižnica stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
</dependency>

<!– Toto je súbor modelov stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
<classifier>models</classifier>
</dependency>

Vytváram statickú objektovú premennú ‘props’, ktorá definuje vlastnosti pre pipeline Stanford Core NLP. Vybral som minimálne vlastnosti, aby bola čo najľahšia. Potom nastavujem anotátory na tokenize, ssplit, pos, parse, sentiment. Vytváram ďalšiu statickú objektovú premennú ‘pipeline’ triedy StanfordCoreNLP. Nakoniec inicializujem pipeline s vlastnosťami ‘props’.

GetSentiment:

Vytvoril som metódu GetSentiment so vstupom typu String a výstupom typu Double. Používam metódu CorrectSpell, ktorú som vytvoril v súbore LanguageCheck.java. Metóda CorrectSpell objektu LanguageCheck mi vracia správny pravopis zadaného tweetu. S týmto opraveným textom používam metódu annotate triedy StanfordCoreNLP. Hodnoty sentimentu, ktoré táto knižnica poskytuje, sú:

0 => veľmi negatívny

1 => negatívny

2 => neutrálny

3 => pozitívny

4 => veľmi pozitívny

Odčítavam od výsledku hodnotu 2 pre každú vetu, aby som získal nasledujúce nové kategórie sentimentu:

-2 => veľmi negatívny

-1 => negatívny

0 => neutrálny

1 => pozitívny

2 => veľmi pozitívny

Vraciam výsledok sentimentu každého tweetu ako priemer sentimentu každej vety tweetu. Tweety nie sú písané v žiadnom štruktúrovanom formáte. Preto nemôžem žiadnemu konkrétnemu riadku tweetu priradiť vyššiu váhu ako ostatným. Predpokladám teda, že každý riadok v tweete má rovnakú dôležitosť. Vraciam premennú ‘total’ typu Double, ktorá obsahuje výslednú hodnotu sentimentu tweetu.

package com.CloudSigma.Spark.TwitterSentiment;

import java.util.List;
import java.util.Properties;
import edu.stanford.nlp.ling.CoreAnnotations.SentencesAnnotation;
import edu.stanford.nlp.neural.rnn.RNNCoreAnnotations;
import edu.stanford.nlp.pipeline.Annotation;
import edu.stanford.nlp.pipeline.StanfordCoreNLP;
import edu.stanford.nlp.sentiment.SentimentCoreAnnotations;
import edu.stanford.nlp.trees.Tree;
import edu.stanford.nlp.util.CoreMap;
import static com.CloudSigma.Spark.TwitterSentiment.LanguageCheck.*;


public class StanfordSentiment  {

	static Properties props = new Properties();
	static {
	props.setProperty("annotators", "tokenize,ssplit,pos,parse,sentiment");}
	static StanfordCoreNLP pipeline = new StanfordCoreNLP(props);


	public static Double GetSentiment(String text) {
		String checkedText = CorrectSpell(text);

		Annotation document = new Annotation(checkedText);
		pipeline.annotate(document);
		List sentences = document.get(SentencesAnnotation.class);

		Double sum = 0.0;

		for (CoreMap sentence : sentences) {
			Tree tree = sentence.get(SentimentCoreAnnotations.SentimentAnnotatedTree.class);
			int sentiment = RNNCoreAnnotations.getPredictedClass(tree);
			int scaled = sentiment - 2;
			sum = sum + scaled;
		}

		Double total = sum / sentences.size();
		System.out.println("Tweet Text: " + checkedText);
		System.out.println("Sentiment Value: " + total);
		return total;
	}

}

balík com.CloudSigma.Spark.TwitterSentiment;

import java.util.List;

import java.util.Properties;

import edu.stanford.nlp.ling.CoreAnnotations.SentencesAnnotation;

import edu.stanford.nlp.neural.rnn.RNNCoreAnnotations;

import edu.stanford.nlp.pipeline.Annotation;

import edu.stanford.nlp.pipeline.StanfordCoreNLP;

import edu.stanford.nlp.sentiment.SentimentCoreAnnotations;

import edu.stanford.nlp.trees.Tree;

import edu.stanford.nlp.util.CoreMap;

import static com.CloudSigma.Spark.TwitterSentiment.LanguageCheck.*;

public class StanfordSentiment {

static Properties props = new Properties();

static {

props.setProperty("annotators", "tokenize,ssplit,pos,parse,sentiment");}

static StanfordCoreNLP pipeline = new StanfordCoreNLP(props);

public static Double GetSentiment(String text) {

String checkedText = CorrectSpell(text);

Annotation document = new Annotation(checkedText);

pipeline.annotate(document);

List sentences = document.get(SentencesAnnotation.class);

Double sum = 0.0;

for (CoreMap sentence : sentences) {

Tree tree = sentence.get(SentimentCoreAnnotations.SentimentAnnotatedTree.class);

int sentiment = RNNCoreAnnotations.getPredictedClass(tree);

int scaled = sentiment - 2;

sum = sum + scaled;

}

Double total = sum / sentences.size();

System.out.println("Text tweetu: " + checkedText);

System.out.println("Hodnota sentimentu: " + total);

return total;

}

Spustenie programu:

Teraz, keď sú tieto dve triedy hotové, pristúpime k ich použitiu. Preto vytváram novú triedu “TwitterDataFlow.java”. Najprv napíšem podmienenú kontrolu, ktorá spustí program iba vtedy, ak je počet odovzdaných vstupných argumentov presne 2. Ak sa počet argumentov nerovná 2, vytlačí správu o nesprávnom použití a tiež sa ukončí s návratovým kódom 1.

Vytváram SparkSession s názvom aplikácie Sentiment Analyzer. Nastavujem vlastnosť hadoop konfigurácie spark kontextu “mapreduce input fileinputformat input dir recursive” na true. To mi umožní rekurzívne načítavať súbory z priečinkov. Vytváram premennú ‘inputPath’ triedy String, do ktorej nastavujem vstupný argument, ako aj ‘/*/*’, čo mi umožní čítať partíciované dáta uložené nástrojom Flume. Čítam json dáta z Flume do Dataset<Row> ‘data’.

Potom registrujem UDF (používateľom definovanú funkciu) v Spark SQL kontexte s názvom ‘Sentiment’, ktorá prijíma String, aplikuje naň metódu GetSentiment triedy StanfordSentiment a vracia dátový typ Double.

Aktuálne mám z flume dáta pre kľúčové slová Apple, Google, Tesla, Infosys, TCS, Oracle, Microsoft a Facebook. Preto vytváram zoznam typu String s týmito kľúčovými slovami. Pre každú z týchto spoločností spúšťam nasledujúce operácie.

Tok operácií:

Najprv vytváram outPath, kam chcem uložiť výsledky. Vytváram dočasný pohľad ‘complete’ nad datasetom ‘data’. Ďalej z dát extrahujem timestamp, partitionBy (na partíciovanie dát pri ukladaní výsledkov), text, main_text (na použitie v regulárnych výrazoch) a followers.

Vytváram dočasný pohľad nad výsledkami a filtrujem z neho dáta konkrétnej spoločnosti. Taktiež aplikujem Sentiment UDF, ktorá mi vracia hodnoty sentimentu v stĺpci ‘seVal’. Serializované dáta uchovávam v pamäti a na disku. Z týchto dát získavam NetSentiment, čo je súčin počtu sledovateľov (Number of Followers) a hodnoty sentimentu (Sentiment Value) daného tweetu. To pomáha určiť vplyv, aký môže mať daný tweet. Na tento účel môžem použiť rôzne vzorce.

Serializované dáta uchovávame v pamäti a na disku, pretože chceme uložiť celý výsledok, keďže analýza sentimentu je výpočtovo náročná úloha. Ak by sme ich neuchovali a plánovali by sme použiť viacero vzorcov na výpočet NetSentimentu alebo vplyvu, v predchádzajúcom dopyte, kde by sme hodnotu metódy Sentiment použili viackrát, by sa analýza sentimentu tweetu vykonávala opakovane.

Zoskupovanie dát:

Teraz zoskupujem dáta podľa stĺpcov timestamp a partitionBy a pre toto zoskupenie počítam priemer NetSentiment. To mi dáva priemerný vplyv spoločnosti, pozitívny alebo negatívny, v konkrétnej minúte. Následne zapisujem výsledky pre každú spoločnosť do outPath, pričom ich rozdeľujem podľa stĺpca partitionBy.

package com.CloudSigma.Spark.TwitterSentiment;

import java.util.Arrays;
import java.util.List;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.storage.StorageLevel;
import static com.CloudSigma.Spark.TwitterSentiment.StanfordSentiment.*;


public class TwitterDataFlow {

	public static void main(String[] args) {

		if (args.length != 2) {
			System.out.println("Nesprávny počet argumentov! \n");
			System.out.println(
					"Použitie: Vstup Výstup \n " + "Vstup: Umiestnenie, odkiaľ sa majú načítať rozdelené dáta"
							+ "Výstup: Kde sa má uložiť konečný výsledok");
			System.exit(1);
		}
		

		

		SparkSession spark = SparkSession.builder().appName("Analyzátor sentimentu").getOrCreate();

		spark.sparkContext().hadoopConfiguration().set("mapreduce.input.fileinputformat.input.dir.recursive", "true");

		String inputPath = args[0] + "/*/*";

		Dataset data = spark.read().json(inputPath);


		spark.sqlContext().udf().register("Sentiment", (String s) -&gt; GetSentiment(s), DataTypes.DoubleType);

		List list = Arrays.asList("apple", "google", "tesla", "infosys", "tcs", "oracle", "microsoft",
				"facebook");

		String query;
		String outPath;

		Dataset result;

		for (String company : list) {

			outPath = args[1] + "/" + company;

			data.createOrReplaceTempView("complete");

			// tmp1 extrahuje TimeStamp, partitionBy (Date), text tweetu, text tweetu malými
			// písmenami a followers_count používateľa, ktorý tweetuje
			Dataset tmp1 = spark.sql(
						"select concat(substr(created_at,5,6), substr(created_at,26,5),' ',substr(created_at,12,6),'00') as timestamp,substr(created_at,5,6) as partitionBy,text,lower(text) as main_text,user.followers_count as followers from complete");

			tmp1.createOrReplaceTempView("tmp");

			// Filtrovanie tweetov, ktoré obsahujú určité názvy spoločností
			Dataset tmp2 = spark.sql("select * from tmp where main_text regexp '(" + company + ")'");

			// Vytvorí pohľad s názvom twitter
			tmp2.createOrReplaceTempView("twitter");

			// tmp3 obsahuje všetky vybrané dáta spolu s hodnotou Sentiment
			// tweetov
			Dataset tmp3 = spark.sql("select  *, Sentiment(text) as seVal from twitter");

			tmp3.persist(StorageLevel.MEMORY_AND_DISK());

			// Vytvorenie ďalšieho pohľadu
			tmp3.createOrReplaceTempView("dataSe");

			Dataset net = spark.sql(
					"select  *,followers*seVal as NetSentiment from dataSe");

			// Vytvorenie finálneho pohľadu na uloženie dát
			net.createOrReplaceTempView("final");

			// Priemerovanie hodnôt Sentiment za minútu zoskupením dát podľa neho
			query = "select timestamp,partitionBy,AVG(NetSentiment) from final group by timestamp,partitionBy";

			// Uloženie výsledku do datasetu result
			result = spark.sql(query);

			// Zápis výsledku na disk
			result.write().partitionBy("partitionBy").json(outPath);
		}
	}
}

package com.CloudSigma.Spark.TwitterSentiment;

import java.util.Arrays;

import java.util.List;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import org.apache.spark.sql.types.DataTypes;

import org.apache.spark.storage.StorageLevel;

import static com.CloudSigma.Spark.TwitterSentiment.StanfordSentiment.*;

public class TwitterDataFlow {

public static void main(String[] args) {

if (args.length != 2) {

System.out.println("Nesprávny počet argumentov! \n");

System.out.println(

"Použitie: Vstup Výstup \n " + "Vstup: Umiestnenie, odkiaľ sa majú načítať rozdelené dáta"

+ "Výstup: Kde sa má uložiť konečný výsledok");

System.exit(1);

}

SparkSession spark = SparkSession.builder().appName("Sentiment Analyzer").getOrCreate();

spark.sparkContext().hadoopConfiguration().set("mapreduce.input.fileinputformat.input.dir.recursive", "true");

String inputPath = args[0] + "/*/*";

Dataset data = spark.read().json(inputPath);

spark.sqlContext().udf().register("Sentiment", (String s) -> GetSentiment(s), DataTypes.DoubleType);

List list = Arrays.asList("apple", "google", "tesla", "infosys", "tcs", "oracle", "microsoft",

"facebook");

String query;

String outPath;

Dataset result;

for (String company : list) {

outPath = args[1] + "/" + company;

data.createOrReplaceTempView("complete");

// tmp1 extrahuje TimeStamp, partitionBy (Dátum), text tweetu, text tweetu v malých

// písmenách a followers_count používateľa, ktorý tweetuje

Dataset tmp1 = spark.sql(

"select concat(substr(created_at,5,6), substr(created_at,26,5),' ',substr(created_at,12,6),'00') as timestamp,substr(created_at,5,6) as partitionBy,text,lower(text) as main_text,user.followers_count as followers from complete");

tmp1.createOrReplaceTempView("tmp");

// Filtrovanie tweetov, ktoré obsahujú určité názvy spoločností

Dataset tmp2 = spark.sql("select * from tmp where main_text regexp '(" + company + ")'");

// Vytvorí pohľad s názvom twitter

tmp2.createOrReplaceTempView("twitter");

// tmp3 obsahuje všetky vybrané dáta spolu s hodnotou Sentimentu

// tweetov

Dataset tmp3 = spark.sql("select *, Sentiment(text) as seVal from twitter");

tmp3.persist(StorageLevel.MEMORY_AND_DISK());

// Vytvorenie ďalšieho pohľadu

tmp3.createOrReplaceTempView("dataSe");

Dataset net = spark.sql(

"select *,followers*seVal as NetSentiment from dataSe");

// Vytvorenie finálneho pohľadu na uloženie dát

net.createOrReplaceTempView("final");

// Priemerovanie hodnôt sentimentu za minútu zoskupením dát podľa nej

query = "select timestamp,partitionBy,AVG(NetSentiment) from final group by timestamp,partitionBy";

// Uloženie výsledku do datasetu result

result = spark.sql(query);

// Zápis výsledku na disk

result.write().partitionBy("partitionBy").json(outPath);

}

Taktiež po dokončení kódu exportujem spustiteľný súbor jar so všetkými závislosťami a skopírujem ho na server, kde chcem túto úlohu spustiť. Následne ju môžem odoslať pomocou nasledujúceho príkazu,

spark-submit --master yarn --deploy-mode cluster SentimentAnalyzer.jar /flume/Twitter/PublicStream/ /flume/output

1	spark-submit --master yarn --deploy-mode cluster SentimentAnalyzer.jar /flume/Twitter/PublicStream/ /flume/output

kde (Zdroj):

--master: master URL pre cluster (napr. spark://23.195.26.187:7077)

--deploy-mode: Či sa má váš ovládač nasadiť na uzly workerov (cluster) alebo lokálne ako externý klient (client) (predvolené: client)
application-jar: Cesta k zabalenému súboru jar vrátane vašej aplikácie a všetkých závislostí. Vezmite do úvahy, že URL adresa musí byť globálne viditeľná vo vašom clustri, napríklad hdfs:// cesta alebo file:// cesta, ktorá je prítomná na všetkých uzloch.
application-arguments: Argumenty odovzdané hlavnej metóde vašej hlavnej triedy, ak nejaké sú. V tomto prípade vstupná a výstupná cesta

Záverečné kroky:

Potom získame výsledky z analýzy sentimentu pomocou Sparku z výstupnej cesty. Napríklad toto je možný výsledok pre apple:

{“timestamp”:”Apr 30 2018 20:31:00″,”avg(NetSentiment)”:-3678.768518518518}
{“timestamp”:”Apr 30 2018 20:32:00″,”avg(NetSentiment)”:-883.002824858757}

Túto aplikáciu som nasadil na CloudSigma s 5-uzlovým HDP clustrom. Konkrétne, každý uzol mal nasledujúcu konfiguráciu:

256 GB SSD
16 GB RAM
20 GHz CPU

Celkovo sa mi podarilo získať výsledky z analýzy sentimentu pomocou Sparku za približne 19 hodín. Okrem toho som zahrnul pokročilejšie výpočty ako program nad dátovou sadou s veľkosťou viac ako 80 GB.

Kód nájdete na GITHUB.

Akshay Nagpal

Autor · CloudSigma

Preslav Dobrev je kreatívny dizajnér v spoločnosti CloudSigma, ktorý sa zameriava na konzistentnú firemnú identitu prostredníctvom tradičných a inovatívnych marketingových kanálov. Dokáže brilantne spájať umeleckú víziu so strategickým marketingom, čím vytvára pôsobivé príbehy značky.

Analýza sentimentu na Twitteri pomocou Spark

Language Tool: Korektor pravopisu

Analyzátor sentimentu: Stanford CoreNLP

GetSentiment:

Spustenie programu:

Tok operácií:

Zoskupovanie dát:

Záverečné kroky:

Akshay Nagpal

Komentáre