In onze vorige post, heb ik een manier uitgewerkt om realtime Twitter-gegevens te extraheren met behulp van Apache Flume. Momenteel heb ik veel gegevens van Twitter verzameld. Daarom zou ik deze willen analyseren en er trends in willen ontdekken. Om sentimentanalyse op de Twitter-gegevens uit te voeren, ga ik een andere Big Data-tool gebruiken, Apache Spark.
Volgens Hortonworks, “Apache Spark is een snelle, in-memory gegevensverwerkingsengine met elegante en expressieve ontwikkelings-API's waarmee data-analisten efficiënt streaming-, machine learning- of SQL-workloads kunnen uitvoeren die snelle iteratieve toegang tot datasets vereisen. Nu Spark op Apache Hadoop YARN draait, kunnen ontwikkelaars overal toepassingen maken om de kracht van Spark te benutten, inzichten te verkrijgen en hun data science-workloads te verrijken binnen een enkele, gedeelde dataset in Hadoop.”
Een Spark-programma kan worden geschreven in JAVA, Scala, Python of R. In dit geval gebruiken we JAVA samen met Maven. Daarnaast wordt Spark geleverd met zowel de HDP- als de Cloudera-distributie. Spark 2 is de huidige versie die wordt gebruikt.
Om de sentimentanalyse met Spark uit te voeren, maak ik een nieuw Maven-project aan. Ik noem het 'Twitter Sentiment Analyzer’. Vervolgens maak ik een klasse aan, “TwitterDataFlow.java”, waarin ik alle vereiste methoden zal implementeren.
Language Tool: Spellingcorrector
In eerste instantie ontdekte ik in de POC dat als de spelling in de tweets onjuist is, de resultaten van de sentimentanalyse nadelig worden beïnvloed. Daarom introduceer ik een SpellChecker. Dit helpt ons om de spelling van de tweets te corrigeren voordat we ze gebruiken voor sentimentanalyse.
Ik ga dit doen door een nieuwe statische methode te maken met de naam ‘CorrectSpell’. Daarnaast ga ik LanguageTool gebruiken om de spelling te controleren en te corrigeren.
Volgens de GIT van LanguageTool’, “LanguageTool is open-source software voor proeflezen voor Engels, Frans, Duits, Pools, Russisch en meer dan 20 andere talen. Het vindt veel fouten die een eenvoudige spellingcontrole niet kan detecteren.”
Vervolgens voeg ik een afhankelijkheid voor de taaltool toe in pom.xml:
<dependency>
<groupId>org.languagetool</groupId>
<artifactId>language-en</artifactId>
<version>4.0</version>
</dependency>
Daarna definieer ik een statische variabele op klasseniveau genaamd langTool van de klasse JLanguageTool. Vervolgens initialiseer ik langTool met een object van de klasse AmericanEnglish.
Vervolgens programmeer ik de methode genaamd SpellChecker met als invoer String text (normale tekst) en als retourtype eveneens String (tekst met gecorrigeerde spelling). Ik gebruik de check-methode van JLanguageTool met de ongecontroleerde tekst als parameter. Vervolgens retourneert deze een lijst met RuleMatch-objecten. Volgens de JLanguageTool Java Docs biedt de klasse RuleMatch “informatie over een foutregel die overeenkomt met tekst en de positie van de overeenkomst.”
Hierna definieer ik drie variabelen: ‘result’ van het type String, ‘lastPos’ van het type integer en ‘tmp’ van het type String. Daarnaast bouw ik bij elke RuleMatch de zin opnieuw op met de eerste voorgestelde spelling van de tool. Daarmee heb ik waar nodig de vereiste try-catch-blokken toegevoegd.
|
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; } } } |
Sentimentanalysator: Stanford CoreNLP
De volgende stap in de sentimentanalyse met Spark is het vinden van sentimenten in de tekst. Om dit te doen, gebruik ik Stanford’s Core NLP Library om sentimentwaarden te vinden. Vervolgens maak ik een klasse genaamd ‘StanfordSentiment’ waarin ik de bibliotheek ga implementeren om de sentimenten binnen onze tekst te vinden.
Om dat te doen, voeg ik de volgende dependencies toe aan het pom.xml-bestand:
<!– Dit is de Stanford Core NLP Library –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
</dependency>
<!– Dit is het modelbestand van Stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
<classifier>models</classifier>
</dependency>
Ik maak een statische objectvariabele aan, ‘props’, die eigenschappen definieert voor de pipeline van Stanford Core NLP. Ik heb de minimale eigenschappen geselecteerd om deze zo licht mogelijk te maken. Daarna stel ik de annotators in op tokenize, ssplit, pos, parse, sentiment. Ik maak nog een statische objectvariabele aan, ‘pipeline’ van de klasse StanfordCoreNLP. Ten slotte initialiseer ik de pipeline met de ‘props’-eigenschappen.
GetSentiment:
Ik heb een methode gemaakt, GetSentiment, met String als invoer en Double als uitvoer. Ik gebruik de methode CorrectSpell die ik in het bestand LanguageCheck.java heb gemaakt. De methode CorrectSpell van het LanguageCheck-object retourneert de correcte spelling van de ingevoerde tweet. Ik gebruik de methode annotate van StanfordCoreNLP met deze gecorrigeerde tekst. De sentimentwaarden die door deze bibliotheek worden gegeven zijn:
0 => zeer negatief
1 => negatief
2 => neutraal
3 => positief
4 => zeer positief
Ik trek 2 van het resultaat af per zin om de volgende nieuwe sentimentcategorieën te krijgen:
-2 => zeer negatief
-1 => negatief
0 => neutraal
1 => positief
2 => zeer positief
Ik retourneer het resultaat van het sentiment van elke tweet als het gemiddelde van het sentiment van elke zin van de tweet. Tweets zijn niet in een gestructureerd formaat geschreven. Daarom kan ik geen specifieke regel van een tweet een hoger gewicht geven dan andere. Daarom neem ik aan dat elke regel in een tweet even belangrijk is. Ik retourneer de variabele ‘total’ van het type Double, die de resulterende sentimentwaarde van de tweet bevat.
|
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 |
pakket 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-tekst: " + checkedText); System.out.println("Sentimentwaarde: " + total); return total; } } |
Programma-initiatie:
Nu deze twee klassen klaar zijn, gaan we verder om ze te gebruiken. Daarom maak ik een nieuwe klasse, “TwitterDataFlow.java”. Eerst schrijf ik een voorwaardelijke controle die het programma alleen uitvoert als het aantal doorgegeven invoerargumenten exact 2 is. Als het aantal argumenten niet gelijk is aan 2, drukt het de melding voor onjuist gebruik af en sluit het ook af met een exitstatus 1.
Ik bouw een SparkSession met de app-naam Sentiment Analyzer. Ik stel de eigenschap van de hadoop-configuratie van de spark-context, “mapreduce input fileinputformat input dir recursive”, in op true. Hiermee kan ik bestanden recursief uit mappen ophalen. Ik maak een variabele ‘inputPath’ van de klasse String waarin ik zowel het invoerargument als ‘/*/*’ instel, waarmee ik de gepartitioneerde gegevens die door Flume zijn opgeslagen kan lezen. Ik lees de json-gegevens van Flume in Dataset<Row> ‘data’.
Vervolgens registreer ik een UDF (User Defined Function) bij de Spark SQL Context, genaamd ‘Sentiment’, die een String aanneemt en de GetSentiment-methode van StanfordSentiment erop toepast en het datatype Double-waarde retourneert.
Momenteel heb ik gegevens van de trefwoorden Apple, Google, Tesla, Infosys, TCS, Oracle, Microsoft en Facebook van Flume. Dus ik maak een lijst van String met deze trefwoorden. Voor elk van deze bedrijven voer ik de volgende bewerkingen uit.
Verloop van de bewerkingen:
Eerst maak ik een outPath aan waar ik de resultaten wil opslaan. Ik maak een temp view, ‘complete’, over de dataset ‘data’. Vervolgens extraheer ik timestamp, partitionBy (om de gegevens te partitioneren tijdens het opslaan van de resultaten), text, main_text (om te gebruiken voor reguliere expressies) en followers uit de gegevens.
Ik maak een temp view over de resultaten en filter de gegevens van een specifiek bedrijf daaruit. Ook pas ik de Sentiment UDF toe, die mij de sentimentwaarden in de kolom ‘seVal’ retourneert. Ik persisteer de geserialiseerde gegevens in het geheugen en als disk spill. Uit deze gegevens haal ik NetSentiment, het product van het aantal volgers en de sentimentwaarde van die tweet. Dit helpt om de invloed te kennen die die tweet kan hebben. Ik kan hiervoor verschillende formules gebruiken.
We persisteren de geserialiseerde gegevens in het geheugen en op de schijf omdat we willen dat het volledige resultaat wordt opgeslagen, aangezien sentimentanalyse een rekenintensieve taak is. Als we dit niet persisteren en we van plan zijn om meerdere formules te gebruiken om de NetSentiment of de invloed te berekenen, zou in de vorige query, waarin we de waarde van de Sentiment-methode meerdere keren zouden gebruiken, de sentimentanalyse van de tweet meerdere keren worden uitgevoerd.
Gegevensgroepering:
Nu groepeer ik de gegevens op timestamp en partitionBy-kolom en bereken ik het gemiddelde van de NetSentiment met deze groepering. Dit geeft me de gemiddelde invloed van het bedrijf, positief of negatief in een specifieke minuut. Vervolgens schrijf ik de resultaten voor elk bedrijf in outPath, waarbij ik het partitioneer op de partitionBy-kolom.
|
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("Onjuist aantal argumenten! \n"); System.out.println( "Gebruik: Input Output \n " + "Input: Locatie waarvandaan de gepartitioneerde gegevens moeten worden gelezen" + "Output: Waar het eindresultaat moet worden opgeslagen"); 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 extraheert TimeStamp, partitionBy (Datum), tweet-tekst, tweet-tekst in kleine // letters en followers_count van de gebruiker die tweet 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"); // Filteren van tweets die bepaalde bedrijfsnamen bevatten Dataset tmp2 = spark.sql("select * from tmp where main_text regexp '(" + company + ")'"); // Maakt een view genaamd twitter tmp2.createOrReplaceTempView("twitter"); // tmp3 bevat alle geselecteerde gegevens samen met de Sentiment-waarde van de // tweets Dataset tmp3 = spark.sql("select *, Sentiment(text) as seVal from twitter"); tmp3.persist(StorageLevel.MEMORY_AND_DISK()); // Nog een view maken tmp3.createOrReplaceTempView("dataSe"); Dataset net = spark.sql( "select *,followers*seVal as NetSentiment from dataSe"); // Een definitieve view maken om de gegevens op te slaan net.createOrReplaceTempView("final"); // Het gemiddelde berekenen van de Sentiment-waarden per minuut door de gegevens erop te groeperen query = "select timestamp,partitionBy,AVG(NetSentiment) from final group by timestamp,partitionBy"; // Het resultaat opslaan in de result-dataset result = spark.sql(query); // Het resultaat naar de schijf schrijven result.write().partitionBy("partitionBy").json(outPath); } } } |
Ook exporteer ik na voltooiing van de code een uitvoerbare jar met alle afhankelijkheden erin en kopieer deze naar de server waar ik deze taak wil uitvoeren. Vervolgens kan ik deze indienen met de volgende opdracht,
|
1 |
spark-submit --master yarn --deploy-mode cluster SentimentAnalyzer.jar /flume/Twitter/PublicStream/ /flume/output |
waarbij (Bron):
--master: De master-URL voor het cluster (bijv.spark://23.195.26.187:7077)
--deploy-mode: Of u uw driver wilt implementeren op de worker-nodes (cluster) of lokaal als een externe client (client) (standaard:client)application-jar: Pad naar een gebundelde jar inclusief uw applicatie en alle afhankelijkheden. Houd er rekening mee dat de URL wereldwijd zichtbaar moet zijn binnen uw cluster, bijvoorbeeld eenhdfs://-pad of eenfile://-pad dat op alle nodes aanwezig is.application-arguments: Argumenten die eventueel aan de main-methode van uw main-klasse worden doorgegeven. Hier, invoer- en uitvoerpad
Laatste stappen:
Vervolgens krijgen we de resultaten van de sentimentanalyse met behulp van Spark uit het uitvoerpad. Dit is bijvoorbeeld een mogelijk resultaat van apple:
{“timestamp”:”Apr 30 2018 20:31:00″,”avg(NetSentiment)”:-3678.768518518518}
{“timestamp”:”Apr 30 2018 20:32:00″,”avg(NetSentiment)”:-883.002824858757}
Ik heb deze applicatie geïmplementeerd op CloudSigma met een 5-node HDP-cluster. Specifiek met de volgende configuratie voor elke node:
256 GB SSD
16 GB RAM
20 GHz CPU
Al met al was ik in staat om in ongeveer 19 uur resultaten te krijgen van de sentimentanalyse met behulp van Spark. Bovendien heb ik geavanceerdere berekeningen opgenomen dan het programma over een dataset van 80+ GB.
De code is te vinden op GITHUB.
Reacties
Nog geen reacties. Wees de eerste.