V našem předchozím příspěvku jsem popsal způsob, jak extrahovat data z Twitteru v reálném čase pomocí Apache Flume. V současné době mám z Twitteru spoustu dat. Proto bych je chtěl analyzovat a najít v nich nějaké trendy. Pro provedení analýzy sentimentu dat z Twitteru použiji další Big Data nástroj, Apache Spark.
Podle Hortonworks, “Apache Spark je rychlý in-memory engine pro zpracování dat s elegantními a expresivními vývojovými API, která datovým analytikům umožňují efektivně spouštět streamování, strojové učení nebo SQL úlohy vyžadující rychlý iterativní přístup k datovým sadám. Se Sparkem běžícím na Apache Hadoop YARN mohou nyní vývojáři kdekoli vytvářet aplikace, které využijí výkon Sparku, získají z nich poznatky a obohatí své úlohy v oblasti datové vědy v rámci jediné sdílené datové sady v Hadoopu.”
Program pro Spark lze napsat v jazyce JAVA, Scala, Python nebo R. V tomto případě budeme používat JAVU společně s Maven. Spark je navíc k dispozici v distribuci HDP i Cloudera. V současnosti se používá verze Spark 2.
Pro provedení analýzy sentimentu pomocí Sparku vytvářím nový Maven projekt. Pojmenovávám ho Twitter Sentiment Analyzer’. Dále vytvářím třídu “TwitterDataFlow.java”, ve které implementuji všechny požadované metody.
Language Tool: Korektor pravopisu
Na začátku jsem v rámci POC zjistil, že pokud je pravopis v tweetech chybný, výsledky analýzy sentimentu jsou nepříznivě ovlivněny. Proto zavádím SpellChecker. Pomůže nám opravit pravopis tweetů předtím, než je použijeme pro analýzu sentimentu.
Udělám to tak, že vytvořím novou statickou metodu s názvem ‘CorrectSpell’. Dále použiji LanguageTool ke kontrole pravopisu a jeho opravě.
Podle LanguageTool’s GIT, “LanguageTool je open-source korektorský software pro angličtinu, francouzštinu, němčinu, polštinu, ruštinu a více než 20 dalších jazyků. Najde mnoho chyb, které jednoduchý korektor pravopisu nedokáže detekovat.”
Dále přidávám závislost pro LanguageTool do pom.xml:
<dependency>
<groupId>org.languagetool</groupId>
<artifactId>language-en</artifactId>
<version>4.0</version>
</dependency>
Poté definuji statickou proměnnou na úrovni třídy langTool třídy JLanguageTool. Následně inicializuji langTool objektem třídy AmericanEnglish.
Dále napíšu metodu s názvem SpellChecker se vstupem typu String text (běžný text) a návratovým typem rovněž String (text se správným pravopisem). Používám metodu check třídy JLanguageTool s parametrem nezkontrolovaného textu. Ta následně vrací seznam objektů RuleMatch. Podle Java dokumentace k JLanguageTool poskytuje třída RuleMatch “informace o chybovém pravidle, které odpovídá textu, a pozici této shody.”
Následně definuji tři proměnné: ‘result’ typu String, ‘lastPos’ typu integer a ‘tmp’ typu String. Navíc u každého RuleMatch znovu sestavuji větu s prvním navrženým pravopisem z nástroje. Tímto jsem také přidal potřebné bloky try-catch všude, kde je to vyžadováno.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
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
Dalším krokem v analýze sentimentu pomocí Sparku je nalezení sentimentu v textu. K tomu používám knihovnu Stanford’s Core NLP, abych zjistil hodnoty sentimentu. Poté vytvořím třídu s názvem ‘StanfordSentiment’, kde tuto knihovnu naimplementuji pro vyhledávání sentimentu v našem textu.
Chcete-li to provést, přidávám do souboru pom.xml následující závislosti:
<!– Toto je knihovna Stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
</dependency>
<!– Toto je soubor modelů Stanford Core NLP’s –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
<classifier>models</classifier>
</dependency>
Vytvářím statickou objektovou proměnnou ‘props’, která definuje vlastnosti pro pipeline Stanford Core NLP’s. Vybral jsem minimální vlastnosti, aby byla co nejlehčí. Poté nastavuji anotátory na tokenize, ssplit, pos, parse, sentiment. Vytvářím další statickou objektovou proměnnou ‘pipeline’ třídy StanfordCoreNLP. Nakonec inicializuji pipeline s vlastnostmi ‘props’.
GetSentiment:
Vytvořil jsem metodu GetSentiment se vstupem typu String a výstupem typu Double. Používám metodu CorrectSpell, kterou jsem vytvořil v souboru LanguageCheck.java. Metoda CorrectSpell objektu LanguageCheck mi vrací správný pravopis zadaného tweetu. S tímto opraveným textem používám metodu annotate třídy StanfordCoreNLP. Hodnoty sentimentu, které tato knihovna poskytuje, jsou:
0 => velmi negativní
1 => negativní
2 => neutrální
3 => pozitivní
4 => velmi pozitivní
Odečítám od výsledku hodnotu 2 pro každou větu, abych získal následující nové kategorie sentimentu:
-2 => velmi negativní
-1 => negativní
0 => neutrální
1 => pozitivní
2 => velmi pozitivní
Vracím výsledek sentimentu každého tweetu jako průměr sentimentu jednotlivých vět tweetu. Tweety nejsou psány v žádném strukturovaném formátu. Proto nemohu žádnému konkrétnímu řádku tweetu přiřadit vyšší váhu než ostatním. Předpokládám tedy, že každý řádek v tweetu má stejný význam. Vracím proměnnou ‘total’ typu Double, která obsahuje výslednou hodnotu sentimentu tweetu.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
balíček 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; } } |
Spuštění programu:
Nyní, když jsou tyto dvě třídy hotové, pokročíme k jejich použití. Proto vytvářím novou třídu “TwitterDataFlow.java”. Nejprve píšu podmíněnou kontrolu, která program spustí pouze tehdy, pokud je počet předaných vstupních argumentů přesně 2. Pokud se počet argumentů nerovná 2, vytiskne zprávu o nesprávném použití a také skončí s návratovým kódem 1.
Vytvářím SparkSession s názvem aplikace Sentiment Analyzer. Nastavuji vlastnost hadoop konfigurace spark context’u “mapreduce input fileinputformat input dir recursive” na true. To mi umožní načítat soubory rekurzivně ze složek. Vytvářím proměnnou ‘inputPath’ třídy String, do které nastavuji vstupní argument a také ‘/*/*’, což mi umožní číst partitionovaná data uložená Flume. Čtu json data z Flume do Dataset<Row> ‘data’.
Poté registruji UDF (User Defined Function) v Spark SQL Contextu s názvem ‘Sentiment’, která přijímá String, aplikuje na něj metodu GetSentiment třídy StanfordSentiment a vrací datový typ Double.
V současné době mám z flume data pro klíčová slova Apple, Google, Tesla, Infosys, TCS, Oracle, Microsoft a Facebook. Vytvářím tedy seznam typu String s těmito klíčovými slovy. Pro každou z těchto společností provádím následující operace.
Průběh operací:
Nejprve vytvářím outPath, kam chci uložit výsledky. Vytvářím dočasný pohled (temp view) ‘complete’ nad datasetem ‘data’. Dále z dat extrahuji timestamp, partitionBy (pro rozdělení dat při ukládání výsledků), text, main_text (pro použití v regulárních výrazech) a followers.
Vytvářím dočasný pohled nad výsledky a filtruji z něj data konkrétní společnosti. Také aplikuji UDF Sentiment, která mi vrací hodnoty sentimentu ve sloupci ‘seVal’. Serializovaná data persistuji v paměti a na disku (disk spill). Z těchto dat získávám NetSentiment, což je součin počtu sledujících (Number of Followers) a hodnoty sentimentu (Sentiment Value) daného tweetu. To pomáhá zjistit vliv, který tento tweet může mít. Pro totéž mohu mít různé vzorce.
Serializovaná data persistujeme v paměti a na disku, protože chceme, aby byl uložen celý výsledek, jelikož analýza sentimentu je výpočetně náročná úloha. Pokud bychom to nepersistovali a plánovali použít více vzorců pro výpočet NetSentimentu nebo vlivu, v předchozím dotazu, kde bychom hodnotu metody Sentiment použili vícekrát, prováděla by se analýza sentimentu tweetu opakovaně.
Seskupování dat:
Nyní seskupuji data podle časového razítka a sloupce partitionBy a průměruji NetSentiment s tímto seskupením. To mi dává průměrný vliv společnosti, pozitivní nebo negativní v konkrétní minutě. Následně zapisuji výsledky pro každou společnost do outPath a rozděluji je podle sloupce partitionBy.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 |
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ávný počet argumentů! \n"); System.out.println( "Použití: Vstup Výstup \n " + "Vstup: Umístění, odkud se mají načíst rozdělená data" + "Výstup: Místo, kam se má uložit konečný výsledek"); 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 (Date), text tweetu, text tweetu v malých // písmenech a followers_count uživatele, který 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"); // Filtrování tweetů obsahujících názvy určitých společností Dataset tmp2 = spark.sql("select * from tmp where main_text regexp '(" + company + ")'"); // Vytvoří pohled s názvem twitter tmp2.createOrReplaceTempView("twitter"); // tmp3 obsahuje všechna vybraná data spolu s hodnotou Sentimentu // tweetů Dataset tmp3 = spark.sql("select *, Sentiment(text) as seVal from twitter"); tmp3.persist(StorageLevel.MEMORY_AND_DISK()); // Vytvoření dalšího pohledu tmp3.createOrReplaceTempView("dataSe"); Dataset net = spark.sql( "select *,followers*seVal as NetSentiment from dataSe"); // Vytvoření finálního pohledu pro uložení dat net.createOrReplaceTempView("final"); // Průměrování hodnot sentimentu za minutu seskupením dat podle ní query = "select timestamp,partitionBy,AVG(NetSentiment) from final group by timestamp,partitionBy"; // Uložení výsledku do datové sady result result = spark.sql(query); // Zápis výsledku na disk result.write().partitionBy("partitionBy").json(outPath); } } } |
Po dokončení kódu také exportuji spustitelný soubor jar se všemi závislostmi a zkopíruji jej na server, kde chci tuto úlohu spustit. Dále ji mohu odeslat pomocí následujícího příkazu,
|
1 |
spark-submit --master yarn --deploy-mode cluster SentimentAnalyzer.jar /flume/Twitter/PublicStream/ /flume/output |
kde (Zdroj):
--master: master URL pro cluster (např.spark://23.195.26.187:7077)
--deploy-mode: Zda nasadit ovladač (driver) na pracovní uzly (cluster) nebo lokálně jako externího klienta (client) (výchozí:client)application-jar: Cesta k zabalenému souboru jar včetně vaší aplikace a všech závislostí. Vezměte v úvahu, že URL musí být globálně viditelná uvnitř vašeho clusteru, napříkladhdfs://cesta nebofile://cesta, která je přítomna na všech uzlech.application-arguments: Argumenty předávané hlavní metodě vaší hlavní třídy, pokud existují. Zde vstupní a výstupní cesta
Závěrečné kroky:
Poté získáme výsledky analýzy sentimentu pomocí Sparku z výstupní cesty. Například toto je možný výsledek pro apple:
{“timestamp”:”Apr 30 2018 20:31:00″,”avg(NetSentiment)”:-3678.768518518518}
{“timestamp”:”Apr 30 2018 20:32:00″,”avg(NetSentiment)”:-883.002824858757}
Tuto aplikaci jsem nasadil na CloudSigma s 5uzlovým HDP clusterem. Konkrétně měl každý uzel následující konfiguraci:
256 GB SSD
16 GB RAM
20 GHz CPU
Celkově se mi podařilo získat výsledky z analýzy sentimentu pomocí Sparku za přibližně 19 hodin. Navíc jsem zahrnul pokročilejší výpočty než program nad datovou sadou o velikosti 80+ GB.
Kód naleznete na GITHUB.
Komentáře
Zatím žádné komentáře. Buďte první.