No nosso post anterior, eu desenvolvi uma maneira de extrair dados do Twitter em tempo real usando o Apache Flume. Atualmente, tenho muitos dados do Twitter. Portanto, eu gostaria de analisá-los e encontrar algumas tendências a partir deles. Para realizar a análise de sentimento dos dados do Twitter, vou usar outra ferramenta de Big Data, o Apache Spark.
De acordo com a Hortonworks, “O Apache Spark é um mecanismo de processamento de dados rápido e em memória com APIs de desenvolvimento elegantes e expressivas para permitir que os profissionais de dados executem com eficiência cargas de trabalho de streaming, machine learning ou SQL que exigem acesso iterativo rápido a conjuntos de dados. Com o Spark rodando no Apache Hadoop YARN, desenvolvedores de todos os lugares podem agora criar aplicativos para explorar o poder do Spark, obter insights e enriquecer suas cargas de trabalho de ciência de dados dentro de um único conjunto de dados compartilhado no Hadoop.”
Um programa Spark pode ser escrito em JAVA, Scala, Python ou R. Neste caso, usaremos JAVA junto com o Maven. Além disso, o Spark vem com as distribuições HDP e Cloudera. O Spark 2 é a versão atual sendo usada.
Para realizar a análise de sentimento com o Spark, estou criando um novo projeto Maven. Estou nomeando-o como Twitter Sentiment Analyzer’. Em seguida, estou criando uma classe, “TwitterDataFlow.java”, na qual implementarei todos os métodos necessários.
Language Tool: Corretor Ortográfico
Inicialmente, na POC, descobri que se a ortografia nos tweets estiver incorreta, os resultados da Análise de Sentimento são afetados negativamente. Portanto, estou introduzindo um SpellChecker. Ele nos ajudará a corrigir a ortografia dos tweets antes de usá-los para a Análise de Sentimento.
Vou fazer isso criando um novo método estático chamado ‘CorrectSpell’. Além disso, vou usar o LanguageTool para verificar as ortografias e corrigi-las.
De acordo com o GIT do LanguageTool, “O LanguageTool é um software de revisão de código aberto para inglês, francês, alemão, polonês, russo e mais de 20 outros idiomas. Ele encontra muitos erros que um corretor ortográfico simples não consegue detectar.”
Em seguida, estou adicionando uma dependência para a ferramenta de idioma no pom.xml:
<dependency>
<groupId>org.languagetool</groupId>
<artifactId>language-en</artifactId>
<version>4.0</version>
</dependency>
Depois disso, estou definindo uma variável estática de nível de classe langTool da classe JLanguageTool. Em seguida, estou inicializando langTool com um objeto da classe AmericanEnglish.
Em seguida, estou codificando o método chamado SpellChecker com a entrada como String text (texto normal) e o tipo de retorno como String (Texto com Ortografia Correta) também. Estou usando o método check do JLanguageTool com o parâmetro como texto não verificado. Em seguida, ele retorna uma lista de RuleMatch. De acordo com o JLanguageTool Java Docs, a classe RuleMatch fornece “informações sobre uma regra de erro que corresponde ao texto e a posição da correspondência.”
Em seguida, estou definindo três variáveis, ‘result’ do tipo String, ‘lastPos’ do tipo inteiro, ‘tmp’ do tipo String. Além disso, com cada RuleMatch, estou recriando a frase com a primeira sugestão de ortografia da ferramenta. Com isso, adicionei os blocos try-catch necessários onde for preciso.
|
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; } } } |
Analisador de Sentimento: Stanford CoreNLP
O próximo passo na análise de sentimento com o Spark é encontrar sentimentos a partir do texto. Para fazer isso, estou usando a Biblioteca Core NLP da Stanford para encontrar valores de sentimento. Em seguida, estou criando uma classe chamada ‘StanfordSentiment’ onde vou implementar a biblioteca para encontrar os sentimentos dentro do nosso texto.
Para fazer isso, estou adicionando as seguintes dependências no arquivo pom.xml:
<!– Esta é a Biblioteca Stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
</dependency>
<!– Este é o arquivo de modelos do Stanford Core NLP –>
<dependency>
<groupId>edu.stanford.nlp</groupId>
<artifactId>stanford-corenlp</artifactId>
<version>3.8.0</version>
<classifier>models</classifier>
</dependency>
Estou criando uma variável de objeto estática, ‘props’, que define as propriedades para o pipeline do Stanford Core NLP. Selecionei as propriedades mínimas para torná-lo o mais leve possível. Depois disso, estou definindo os anotadores para tokenize, ssplit, pos, parse, sentiment. Estou criando outra variável de objeto estática, ‘pipeline’, da classe StanfordCoreNLP. Finalmente, inicializo o pipeline com as propriedades ‘props’.
GetSentiment:
Criei um método, GetSentiment, com entrada como String e saída como Double. Estou usando o método CorrectSpell que criei no arquivo LanguageCheck.java. O método CorrectSpell do objeto LanguageCheck me retorna a ortografia correta do tweet inserido. Eu uso o método annotate do StanfordCoreNLP com este texto corrigido. Os valores de sentimento fornecidos por esta biblioteca são:
0 => muito negativo
1 => negativo
2 => neutro
3 => positivo
4 => muito positivo
Estou subtraindo o resultado por 2 por frase para obter as seguintes novas categorias de Sentimento:
-2 => muito negativo
-1 => negativo
0 => neutro
1 => positivo
2 => muito positivo
Estou retornando o resultado do sentimento de cada tweet como a média do sentimento de cada frase do tweet. Os tweets não são escritos em nenhum formato estruturado. Sendo assim, não posso atribuir a nenhuma linha específica de um tweet um peso maior do que a outras. Portanto, estou assumindo que cada linha em um tweet tem igual importância. Estou retornando a variável ‘total’ do tipo Double, que possui o valor de sentimento resultante do tweet.
|
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 |
pacote 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("Texto do Tweet: " + checkedText); System.out.println("Valor do Sentimento: " + total); return total; } } |
Inicialização do Programa:
Agora que essas duas classes estão prontas, vamos prosseguir para usá-las. Assim, estou criando uma nova classe, “TwitterDataFlow.java”. Primeiro, estou escrevendo uma verificação condicional que só executará o programa se o número de argumentos de entrada passados for exatamente 2. Se o número de argumentos não for igual a 2, ele imprimirá a mensagem de uso incorreto e também sairá com um status de saída 1.
Estou construindo uma SparkSession com o nome do aplicativo como Sentiment Analyzer. Estou definindo a propriedade de configuração do hadoop do spark context, “mapreduce input fileinputformat input dir recursive” como true. Isso me permitirá recuperar arquivos recursivamente de pastas. Estou criando uma variável, ‘inputPath’ da classe String na qual estou definindo o argumento de entrada, bem como ‘/*/*’, o que me permitirá ler os dados particionados armazenados pelo Flume. Estou lendo os dados json do Flume no Dataset<Row> ‘data’.
Em seguida, estou registrando uma UDF (User Defined Function) com o Spark SQL Context, chamada ‘Sentiment’, que recebe uma String e aplica o método GetSentiment do StanfordSentiment sobre ela e retorna o tipo de dados de valor Double.
Atualmente, tenho dados das palavras-chave Apple, Google, Tesla, Infosys, TCS, Oracle, Microsoft e Facebook do flume. Portanto, estou criando uma lista de String com essas palavras-chave. Para cada uma dessas empresas, estou executando as seguintes operações.
Fluxo de operações:
Primeiro, estou criando um outPath onde quero salvar os resultados. Estou criando uma temp view, ‘complete’ sobre o dataset, ‘data’. Em seguida, estou extraindo timestamp, partitionBy (para particionar os dados ao armazenar os resultados), text, main_text (para usar em expressões regulares), followers a partir dos dados.
Estou criando uma temp view sobre os resultados e filtrando os dados de uma empresa específica a partir disso. Além disso, estou aplicando a UDF Sentiment, que me retorna os valores de sentimento na coluna ‘seVal’. Estou persistindo os dados serializados na memória e como gravação em disco (disk spill). A partir desses dados, estou obtendo o NetSentiment, o produto do Número de Seguidores e o Valor de Sentimento daquele tweet. Isso ajuda a conhecer a influência que o tweet pode ter. Posso ter fórmulas diferentes para o mesmo.
Estamos persistindo os dados serializados na memória e no disco porque queremos que todo o resultado seja armazenado, já que a análise de sentimento é uma tarefa computacionalmente pesada. Se não persistirmos isso e planejarmos usar várias fórmulas para calcular o NetSentiment ou a influência, na consulta anterior onde usaríamos o valor do método Sentiment várias vezes, ela faria a Análise de Sentimento do tweet várias vezes.
Agrupamento de dados:
Agora, estou agrupando os dados por timestamp e pela coluna partitionBy e calculando a média do NetSentiment com este agrupamento. Isso me dá a influência média da empresa, positiva ou negativa em um minuto específico. Consequentemente, estou gravando os resultados de cada empresa no outPath, particionando-o pela coluna 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("Número incorreto de argumentos! \n"); System.out.println( "Uso: Input Output \n " + "Input: Local de onde os dados particionados precisam ser lidos" + "Output: Onde o resultado final precisa ser armazenado"); System.exit(1); } SparkSession spark = SparkSession.builder().appName("Analisador de Sentimento").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 extrai TimeStamp, partitionBy (Date), texto do tweet, texto do tweet em minúsculas // e followers_count do usuário que tuitou 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"); // Filtrando tweets que contêm determinados nomes de empresas Dataset tmp2 = spark.sql("select * from tmp where main_text regexp '(" + company + ")'"); // Cria uma view chamada twitter tmp2.createOrReplaceTempView("twitter"); // tmp3 contém todos os dados selecionados junto com o valor de sentimento dos // tweets Dataset tmp3 = spark.sql("select *, Sentiment(text) as seVal from twitter"); tmp3.persist(StorageLevel.MEMORY_AND_DISK()); // Criando outra view tmp3.createOrReplaceTempView("dataSe"); Dataset net = spark.sql( "select *,followers*seVal as NetSentiment from dataSe"); // Criando uma view final para salvar os dados net.createOrReplaceTempView("final"); // Calculando a média dos valores de sentimento por minuto agrupando os dados por ele query = "select timestamp,partitionBy,AVG(NetSentiment) from final group by timestamp,partitionBy"; // Salvando o resultado no dataset result result = spark.sql(query); // Gravando o resultado no disco result.write().partitionBy("partitionBy").json(outPath); } } } |
Além disso, após a conclusão do código, exporto um jar executável com todas as dependências nele e o copio para o servidor, onde quero executar este job. Em seguida, posso enviá-lo usando o seguinte comando,
|
1 |
spark-submit --master yarn --deploy-mode cluster SentimentAnalyzer.jar /flume/Twitter/PublicStream/ /flume/output |
onde (Origem):
--master: A URL master para o cluster (ex:spark://23.195.26.187:7077)
--deploy-mode: Se deve implantar seu driver nos nós de trabalho (cluster) ou localmente como um cliente externo (client) (padrão:client)application-jar: Caminho para um jar empacotado incluindo sua aplicação e todas as dependências. Leve em consideração que a URL deve ser visível globalmente dentro do seu cluster, por exemplo, um caminhohdfs://ou um caminhofile://que esteja presente em todos os nós.application-arguments: Argumentos passados para o método principal da sua classe principal, se houver. Aqui, caminho de entrada e saída
Passos Finais:
Em seguida, obteremos os resultados da análise de sentimento usando o Spark a partir do caminho de saída. Por exemplo, este é um resultado possível para a apple:
{“timestamp”:”Apr 30 2018 20:31:00″,”avg(NetSentiment)”:-3678.768518518518}
{“timestamp”:”Apr 30 2018 20:32:00″,”avg(NetSentiment)”:-883.002824858757}
Eu implantei esta aplicação na CloudSigma com um cluster HDP de 5 nós. Especificamente, com cada nó tendo a seguinte configuração:
256 GB SSD
16 GB RAM
20 GHz CPU
No geral, consegui obter resultados da análise de sentimento usando o Spark em aproximadamente 19 horas. Além disso, incluí cálculos mais avançados do que o programa em um conjunto de dados de mais de 80 GB.
O código pode ser encontrado no GITHUB.
Comentários
Nenhum comentário ainda. Seja o primeiro.