Apache Spark - Guia rápido

As indústrias estão usando o Hadoop extensivamente para analisar seus conjuntos de dados. O motivo é que a estrutura do Hadoop é baseada em um modelo de programação simples (MapReduce) e permite uma solução de computação que é escalável, flexível, tolerante a falhas e econômica. Aqui, a principal preocupação é manter a velocidade no processamento de grandes conjuntos de dados em termos de tempo de espera entre as consultas e tempo de espera para executar o programa.

Spark foi apresentado pela Apache Software Foundation para acelerar o processo de software de computação computacional Hadoop.

Contra uma crença comum, Spark is not a modified version of Hadoope não é, realmente, dependente do Hadoop porque tem seu próprio gerenciamento de cluster. Hadoop é apenas uma das maneiras de implementar o Spark.

O Spark usa o Hadoop de duas maneiras - uma é storage e o segundo é processing. Como o Spark tem sua própria computação de gerenciamento de cluster, ele usa o Hadoop apenas para fins de armazenamento.

Apache Spark

Apache Spark é uma tecnologia de computação em cluster ultrarrápida, projetada para computação rápida. É baseado no Hadoop MapReduce e estende o modelo MapReduce para usá-lo com eficiência para mais tipos de cálculos, o que inclui consultas interativas e processamento de fluxo. A principal característica do Spark é o seuin-memory cluster computing que aumenta a velocidade de processamento de um aplicativo.

O Spark foi projetado para cobrir uma ampla gama de cargas de trabalho, como aplicativos em lote, algoritmos iterativos, consultas interativas e streaming. Além de suportar toda essa carga de trabalho em um respectivo sistema, ele reduz a carga de gerenciamento de manter ferramentas separadas.

Evolução do Apache Spark

Spark é um dos subprojetos do Hadoop desenvolvido em 2009 no AMPLab da UC Berkeley por Matei Zaharia. Foi Open Sourced em 2010 sob uma licença BSD. Ele foi doado à fundação de software Apache em 2013 e agora o Apache Spark se tornou um projeto Apache de nível superior a partir de fevereiro de 2014.

Recursos do Apache Spark

O Apache Spark possui os seguintes recursos.

Speed- O Spark ajuda a executar um aplicativo no cluster Hadoop, até 100 vezes mais rápido na memória e 10 vezes mais rápido quando executado no disco. Isso é possível reduzindo o número de operações de leitura / gravação no disco. Ele armazena os dados de processamento intermediários na memória.
Supports multiple languages- Spark fornece APIs integradas em Java, Scala ou Python. Portanto, você pode escrever aplicativos em diferentes idiomas. O Spark oferece 80 operadores de alto nível para consultas interativas.
Advanced Analytics- O Spark não suporta apenas 'Mapear' e 'reduzir'. Ele também oferece suporte a consultas SQL, dados de streaming, aprendizado de máquina (ML) e algoritmos de gráfico.

Spark construído em Hadoop

O diagrama a seguir mostra três maneiras de como o Spark pode ser construído com componentes do Hadoop.

Existem três maneiras de implantação do Spark, conforme explicado abaixo.

Standalone- A implantação autônoma do Spark significa que o Spark ocupa o lugar acima do HDFS (Hadoop Distributed File System) e o espaço é alocado para o HDFS, explicitamente. Aqui, o Spark e o MapReduce serão executados lado a lado para cobrir todos os jobs do Spark no cluster.
Hadoop Yarn- Implementação do Hadoop Yarn significa, simplesmente, que o Spark roda no Yarn sem qualquer pré-instalação ou acesso root necessário. Isso ajuda a integrar o Spark ao ecossistema Hadoop ou pilha Hadoop. Ele permite que outros componentes sejam executados no topo da pilha.
Spark in MapReduce (SIMR)- Spark no MapReduce é usado para iniciar o trabalho de faísca além da implantação autônoma. Com o SIMR, o usuário pode iniciar o Spark e usar seu shell sem qualquer acesso administrativo.

Componentes do Spark

A ilustração a seguir descreve os diferentes componentes do Spark.

Apache Spark Core

Spark Core é o mecanismo de execução geral subjacente para a plataforma Spark sobre a qual todas as outras funcionalidades são construídas. Ele fornece computação In-Memory e conjuntos de dados de referência em sistemas de armazenamento externo.

Spark SQL

Spark SQL é um componente no topo do Spark Core que apresenta uma nova abstração de dados chamada SchemaRDD, que fornece suporte para dados estruturados e semiestruturados.

Spark Streaming

O Spark Streaming aproveita a capacidade de agendamento rápido do Spark Core para realizar análises de streaming. Ele ingere dados em minilotes e realiza transformações RDD (conjuntos de dados distribuídos resilientes) nesses minilotes de dados.

MLlib (Biblioteca de aprendizado de máquina)

MLlib é uma estrutura de aprendizado de máquina distribuída acima do Spark por causa da arquitetura do Spark baseada em memória distribuída. É, de acordo com benchmarks, feito pelos desenvolvedores MLlib contra as implementações Alternating Least Squares (ALS). Spark MLlib é nove vezes mais rápido que a versão baseada em disco do Hadoop doApache Mahout (antes de Mahout ganhar uma interface Spark).

GraphX

GraphX é uma estrutura de processamento de gráfico distribuída sobre o Spark. Ele fornece uma API para expressar computação de gráfico que pode modelar os gráficos definidos pelo usuário usando a API de abstração Pregel. Ele também fornece um tempo de execução otimizado para essa abstração.

Conjuntos de dados distribuídos resilientes

Resilient Distributed Datasets (RDD) é uma estrutura de dados fundamental do Spark. É uma coleção imutável de objetos distribuídos. Cada conjunto de dados em RDD é dividido em partições lógicas, que podem ser computadas em diferentes nós do cluster. Os RDDs podem conter qualquer tipo de objetos Python, Java ou Scala, incluindo classes definidas pelo usuário.

Formalmente, um RDD é uma coleção de registros particionada somente leitura. Os RDDs podem ser criados por meio de operações determinísticas em dados no armazenamento estável ou em outros RDDs. RDD é uma coleção de elementos tolerantes a falhas que podem ser operados em paralelo.

Existem duas maneiras de criar RDDs - parallelizing uma coleção existente em seu programa de driver, ou referencing a dataset em um sistema de armazenamento externo, como um sistema de arquivo compartilhado, HDFS, HBase ou qualquer fonte de dados que ofereça um formato de entrada Hadoop.

O Spark usa o conceito de RDD para obter operações MapReduce mais rápidas e eficientes. Vamos primeiro discutir como as operações MapReduce ocorrem e por que não são tão eficientes.

O compartilhamento de dados é lento no MapReduce

MapReduce é amplamente adotado para processar e gerar grandes conjuntos de dados com um algoritmo paralelo distribuído em um cluster. Ele permite aos usuários escrever cálculos paralelos, usando um conjunto de operadores de alto nível, sem ter que se preocupar com a distribuição do trabalho e tolerância a falhas.

Infelizmente, na maioria das estruturas atuais, a única maneira de reutilizar dados entre cálculos (Ex - entre duas tarefas MapReduce) é gravá-los em um sistema de armazenamento externo estável (Ex - HDFS). Embora essa estrutura forneça várias abstrações para acessar os recursos computacionais de um cluster, os usuários ainda querem mais.

Ambos Iterative e Interactiveos aplicativos exigem compartilhamento de dados mais rápido em trabalhos paralelos. O compartilhamento de dados é lento no MapReduce devido areplication, serializatione disk IO. Com relação ao sistema de armazenamento, a maioria dos aplicativos Hadoop passam mais de 90% do tempo executando operações de leitura e gravação HDFS.

Operações iterativas no MapReduce

Reutilize resultados intermediários em vários cálculos em aplicativos de vários estágios. A ilustração a seguir explica como a estrutura atual funciona, ao fazer as operações iterativas no MapReduce. Isso gera sobrecargas substanciais devido à replicação de dados, E / S de disco e serialização, o que torna o sistema lento.

Operações interativas no MapReduce

O usuário executa consultas ad-hoc no mesmo subconjunto de dados. Cada consulta fará a E / S do disco no armazenamento estável, que pode dominar o tempo de execução do aplicativo.

A ilustração a seguir explica como a estrutura atual funciona ao fazer as consultas interativas no MapReduce.

Compartilhamento de dados usando Spark RDD

O compartilhamento de dados é lento no MapReduce devido a replication, serializatione disk IO. A maioria dos aplicativos Hadoop, eles passam mais de 90% do tempo fazendo operações de leitura e gravação HDFS.

Reconhecendo esse problema, os pesquisadores desenvolveram uma estrutura especializada chamada Apache Spark. A ideia chave da faísca éResiliente Distribuído Datasets (RDD); ele suporta computação de processamento na memória. Isso significa que ele armazena o estado da memória como um objeto entre os trabalhos e o objeto pode ser compartilhado entre esses trabalhos. O compartilhamento de dados na memória é de 10 a 100 vezes mais rápido que a rede e o disco.

Vamos agora tentar descobrir como as operações iterativas e interativas ocorrem no Spark RDD.

Operações iterativas no Spark RDD

A ilustração a seguir mostra as operações iterativas no Spark RDD. Ele armazenará resultados intermediários em uma memória distribuída em vez de armazenamento estável (disco) e tornará o sistema mais rápido.

Note - Se a memória distribuída (RAM) for suficiente para armazenar os resultados intermediários (Estado do JOB), ela armazenará esses resultados no disco.

Operações interativas no Spark RDD

Esta ilustração mostra operações interativas no Spark RDD. Se diferentes consultas são executadas no mesmo conjunto de dados repetidamente, esses dados específicos podem ser mantidos na memória para melhores tempos de execução.

Por padrão, cada RDD transformado pode ser recalculado cada vez que você executa uma ação nele. No entanto, você também podepersistum RDD na memória, caso em que o Spark manterá os elementos no cluster para um acesso muito mais rápido na próxima vez que você o consultar. Também há suporte para RDDs persistentes no disco ou replicados em vários nós.

Spark é o subprojeto do Hadoop. Portanto, é melhor instalar o Spark em um sistema baseado em Linux. As etapas a seguir mostram como instalar o Apache Spark.

Etapa 1: Verificar a instalação do Java

A instalação do Java é uma das coisas obrigatórias na instalação do Spark. Tente o seguinte comando para verificar a versão JAVA.

$java -version

Se o Java já estiver instalado em seu sistema, você verá a seguinte resposta -

java version "1.7.0_71" 
Java(TM) SE Runtime Environment (build 1.7.0_71-b13) 
Java HotSpot(TM) Client VM (build 25.0-b02, mixed mode)

Caso você não tenha o Java instalado em seu sistema, instale o Java antes de prosseguir para a próxima etapa.

Etapa 2: Verificando a instalação do Scala

Você deve usar a linguagem Scala para implementar o Spark. Portanto, vamos verificar a instalação do Scala usando o seguinte comando.

$scala -version

Se o Scala já estiver instalado em seu sistema, você verá a seguinte resposta -

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

Caso você não tenha o Scala instalado em seu sistema, prossiga para a próxima etapa para a instalação do Scala.

Etapa 3: Baixando Scala

Baixe a versão mais recente do Scala, visite o seguinte link Baixar Scala . Para este tutorial, estamos usando a versão scala-2.11.6. Após o download, você encontrará o arquivo Scala tar na pasta de download.

Etapa 4: Instalando o Scala

Siga as etapas fornecidas abaixo para instalar o Scala.

Extraia o arquivo Scala tar

Digite o seguinte comando para extrair o arquivo tar do Scala.

$ tar xvf scala-2.11.6.tgz

Mover arquivos de software Scala

Use os seguintes comandos para mover os arquivos do software Scala para o respectivo diretório (/usr/local/scala).

$ su – 
Password: 
# cd /home/Hadoop/Downloads/ 
# mv scala-2.11.6 /usr/local/scala 
# exit

Definir PATH para Scala

Use o seguinte comando para definir PATH para Scala.

$ export PATH = $PATH:/usr/local/scala/bin

Verificando a instalação do Scala

Após a instalação, é melhor verificar isso. Use o seguinte comando para verificar a instalação do Scala.

$scala -version

Se o Scala já estiver instalado em seu sistema, você verá a seguinte resposta -

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

Etapa 5: baixando o Apache Spark

Baixe a versão mais recente do Spark visitando o seguinte link: Baixe o Spark . Para este tutorial, estamos usandospark-1.3.1-bin-hadoop2.6versão. Depois de baixá-lo, você encontrará o arquivo Spark tar na pasta de download.

Etapa 6: Instalando o Spark

Siga as etapas abaixo para instalar o Spark.

Extraindo piche

O seguinte comando para extrair o arquivo spark tar.

$ tar xvf spark-1.3.1-bin-hadoop2.6.tgz

Movendo arquivos de software Spark

Os seguintes comandos para mover os arquivos do software Spark para o respectivo diretório (/usr/local/spark).

$ su – 
Password:  

# cd /home/Hadoop/Downloads/ 
# mv spark-1.3.1-bin-hadoop2.6 /usr/local/spark 
# exit

Configurando o ambiente para Spark

Adicione a seguinte linha a ~/.bashrcArquivo. Isso significa adicionar o local onde o arquivo do software spark está localizado na variável PATH.

export PATH=$PATH:/usr/local/spark/bin

Use o seguinte comando para obter o arquivo ~ / .bashrc.

$ source ~/.bashrc

Etapa 7: Verificando a instalação do Spark

Escreva o seguinte comando para abrir o shell do Spark.

$spark-shell

Se o Spark for instalado com sucesso, você encontrará a seguinte saída.

Spark assembly has been built with Hive, including Datanucleus jars on classpath 
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties 
15/06/04 15:25:22 INFO SecurityManager: Changing view acls to: hadoop 
15/06/04 15:25:22 INFO SecurityManager: Changing modify acls to: hadoop
15/06/04 15:25:22 INFO SecurityManager: SecurityManager: authentication disabled;
   ui acls disabled; users with view permissions: Set(hadoop); users with modify permissions: Set(hadoop) 
15/06/04 15:25:22 INFO HttpServer: Starting HTTP Server 
15/06/04 15:25:23 INFO Utils: Successfully started service 'HTTP class server' on port 43292. 
Welcome to 
      ____              __ 
     / __/__  ___ _____/ /__ 
    _\ \/ _ \/ _ `/ __/  '_/ 
   /___/ .__/\_,_/_/ /_/\_\   version 1.4.0 
      /_/  
		
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71) 
Type in expressions to have them evaluated. 
Spark context available as sc  
scala>

Spark Core é a base de todo o projeto. Ele fornece despacho distribuído de tarefas, agendamento e funcionalidades básicas de E / S. O Spark usa uma estrutura de dados fundamental especializada conhecida como RDD (Resilient Distributed Datasets), que é uma coleção lógica de dados particionados entre máquinas. Os RDDs podem ser criados de duas maneiras; um é fazendo referência a conjuntos de dados em sistemas de armazenamento externo e, segundo, aplicando transformações (por exemplo, mapa, filtro, redutor, junção) em RDDs existentes.

A abstração RDD é exposta por meio de uma API integrada à linguagem. Isso simplifica a complexidade da programação porque a maneira como os aplicativos manipulam RDDs é semelhante à manipulação de coleções locais de dados.

Spark Shell

O Spark fornece um shell interativo - uma ferramenta poderosa para analisar dados interativamente. Ele está disponível em linguagem Scala ou Python. A abstração primária do Spark é uma coleção distribuída de itens chamada Resilient Distributed Dataset (RDD). Os RDDs podem ser criados a partir de formatos de entrada do Hadoop (como arquivos HDFS) ou transformando outros RDDs.

Abra o Spark Shell

O seguinte comando é usado para abrir o shell do Spark.

$ spark-shell

Crie um RDD simples

Vamos criar um RDD simples a partir do arquivo de texto. Use o seguinte comando para criar um RDD simples.

scala> val inputfile = sc.textFile(“input.txt”)

A saída para o comando acima é

inputfile: org.apache.spark.rdd.RDD[String] = input.txt MappedRDD[1] at textFile at <console>:12

A API Spark RDD apresenta alguns Transformations e poucos Actions para manipular RDD.

Transformações RDD

As transformações RDD retornam o ponteiro para o novo RDD e permitem que você crie dependências entre os RDDs. Cada RDD na cadeia de dependência (String de Dependências) possui uma função para calcular seus dados e possui um ponteiro (dependência) para seu RDD pai.

O Spark é preguiçoso, então nada será executado a menos que você chame alguma transformação ou ação que acione a criação e execução do job. Observe o seguinte trecho do exemplo de contagem de palavras.

Portanto, a transformação RDD não é um conjunto de dados, mas uma etapa em um programa (pode ser a única etapa) dizendo ao Spark como obter dados e o que fazer com eles.

A seguir está uma lista de transformações RDD.

S.No	Transformações e significado
1	map(func) Retorna um novo conjunto de dados distribuído, formado pela passagem de cada elemento da fonte por meio de uma função func.
2	filter(func) Retorna um novo conjunto de dados formado selecionando os elementos da fonte em que func retorna verdadeiro.
3	flatMap(func) Semelhante ao map, mas cada item de entrada pode ser mapeado para 0 ou mais itens de saída (portanto, func deve retornar um Seq em vez de um único item).
4	mapPartitions(func) Semelhante ao map, mas é executado separadamente em cada partição (bloco) do RDD, então func deve ser do tipo Iterator <T> ⇒ Iterator <U> quando executado em um RDD do tipo T.
5	mapPartitionsWithIndex(func) Semelhante às partições de mapas, mas também fornece func com um valor inteiro representando o índice da partição, então func deve ser do tipo (Int, Iterator <T>) ⇒ Iterator <U> ao executar em um RDD do tipo T.
6	sample(withReplacement, fraction, seed) Experimente um fraction dos dados, com ou sem substituição, usando uma determinada semente do gerador de número aleatório.
7	union(otherDataset) Retorna um novo conjunto de dados que contém a união dos elementos no conjunto de dados de origem e o argumento.
8	intersection(otherDataset) Retorna um novo RDD que contém a interseção de elementos no conjunto de dados de origem e o argumento.
9	distinct([numTasks]) Retorna um novo conjunto de dados que contém os elementos distintos do conjunto de dados de origem.
10	groupByKey([numTasks]) Quando chamado em um conjunto de dados de pares (K, V), retorna um conjunto de dados de pares (K, Iteráveis <V>). Note - Se você estiver agrupando para realizar uma agregação (como uma soma ou média) em cada chave, o uso de reduceByKey ou aggregateByKey resultará em um desempenho muito melhor.
11	reduceByKey(func, [numTasks]) Quando chamado em um conjunto de dados de pares (K, V), retorna um conjunto de dados de pares (K, V) onde os valores de cada chave são agregados usando a função de redução fornecida func , que deve ser do tipo (V, V) ⇒ V Como em groupByKey, o número de tarefas de redução é configurável por meio de um segundo argumento opcional.
12	aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) Quando chamado em um conjunto de dados de pares (K, V), retorna um conjunto de dados de pares (K, U) onde os valores de cada chave são agregados usando as funções de combinação fornecidas e um valor neutro "zero". Permite um tipo de valor agregado diferente do tipo de valor de entrada, evitando alocações desnecessárias. Como em groupByKey, o número de tarefas de redução é configurável por meio de um segundo argumento opcional.
13	sortByKey([ascending], [numTasks]) Quando chamado em um conjunto de dados de pares (K, V) onde K implementa Ordered, retorna um conjunto de dados de pares (K, V) classificados por chaves em ordem crescente ou decrescente, conforme especificado no argumento ascendente booleano.
14	join(otherDataset, [numTasks]) Quando chamado em conjuntos de dados do tipo (K, V) e (K, W), retorna um conjunto de dados de pares (K, (V, W)) com todos os pares de elementos para cada chave. As junções externas são suportadas por leftOuterJoin, rightOuterJoin e fullOuterJoin.
15	cogroup(otherDataset, [numTasks]) Quando chamado em conjuntos de dados do tipo (K, V) e (K, W), retorna um conjunto de dados de tuplas (K, (Iterable <V>, Iterable <W>)). Esta operação também é chamada de grupo com.
16	cartesian(otherDataset) Quando chamado em conjuntos de dados dos tipos T e U, retorna um conjunto de dados de pares (T, U) (todos os pares de elementos).
17	pipe(command, [envVars]) Canalize cada partição do RDD por meio de um comando shell, por exemplo, um script Perl ou bash. Os elementos RDD são gravados no stdin do processo e as linhas de saída do stdout são retornadas como um RDD de strings.
18	coalesce(numPartitions) Diminua o número de partições no RDD para numPartitions. Útil para executar operações com mais eficiência após filtrar um grande conjunto de dados.
19	repartition(numPartitions) Reorganize os dados no RDD aleatoriamente para criar mais ou menos partições e equilibrá-los entre elas. Isso sempre embaralha todos os dados pela rede.
20	repartitionAndSortWithinPartitions(partitioner) Reparticione o RDD de acordo com o particionador fornecido e, dentro de cada partição resultante, classifique os registros por suas chaves. Isso é mais eficiente do que chamar a repartição e, em seguida, classificar dentro de cada partição, porque pode empurrar a classificação para o mecanismo de embaralhamento.

Ações

A tabela a seguir fornece uma lista de ações, que retornam valores.

S.No	Ação e Significado
1	reduce(func) Agregue os elementos do conjunto de dados usando uma função func(que recebe dois argumentos e retorna um). A função deve ser comutativa e associativa para que possa ser calculada corretamente em paralelo.
2	collect() Retorna todos os elementos do conjunto de dados como uma matriz no programa de driver. Isso geralmente é útil após um filtro ou outra operação que retorna um subconjunto suficientemente pequeno de dados.
3	count() Retorna o número de elementos no conjunto de dados.
4	first() Retorna o primeiro elemento do conjunto de dados (semelhante a take (1)).
5	take(n) Retorna uma matriz com o primeiro n elementos do conjunto de dados.
6	takeSample (withReplacement,num, [seed]) Retorna uma matriz com uma amostra aleatória de num elementos do conjunto de dados, com ou sem substituição, opcionalmente pré-especificando uma semente do gerador de número aleatório.
7	takeOrdered(n, [ordering]) Retorna o primeiro n elementos do RDD usando sua ordem natural ou um comparador personalizado.
8	saveAsTextFile(path) Grava os elementos do conjunto de dados como um arquivo de texto (ou conjunto de arquivos de texto) em um determinado diretório no sistema de arquivos local, HDFS ou qualquer outro sistema de arquivos compatível com Hadoop. Spark chama toString em cada elemento para convertê-lo em uma linha de texto no arquivo.
9	saveAsSequenceFile(path) (Java and Scala) Grava os elementos do conjunto de dados como um Hadoop SequenceFile em um determinado caminho no sistema de arquivos local, HDFS ou qualquer outro sistema de arquivos compatível com Hadoop. Isso está disponível em RDDs de pares chave-valor que implementam a interface gravável do Hadoop. No Scala, ele também está disponível em tipos que são implicitamente conversíveis em graváveis (o Spark inclui conversões para tipos básicos como Int, Double, String, etc.).
10	saveAsObjectFile(path) (Java and Scala) Grava os elementos do conjunto de dados em um formato simples usando serialização Java, que pode então ser carregado usando SparkContext.objectFile ().
11	countByKey() Disponível apenas em RDDs do tipo (K, V). Retorna um hashmap de pares (K, Int) com a contagem de cada chave.
12	foreach(func) Executa uma função funcem cada elemento do conjunto de dados. Isso geralmente é feito para efeitos colaterais, como atualizar um acumulador ou interagir com sistemas de armazenamento externo. Note- modificar variáveis diferentes de Accumulators fora de foreach () pode resultar em comportamento indefinido. Consulte Noções básicas sobre fechamentos para obter mais detalhes.

Programação com RDD

Vejamos as implementações de algumas transformações e ações RDD na programação RDD com a ajuda de um exemplo.

Exemplo

Considere um exemplo de contagem de palavras - conta cada palavra que aparece em um documento. Considere o seguinte texto como uma entrada e é salvo como uminput.txt arquivo em um diretório inicial.

input.txt - arquivo de entrada.

people are not as beautiful as they look, 
as they walk or as they talk.
they are only as beautiful  as they love, 
as they care as they share.

Siga o procedimento fornecido a seguir para executar o exemplo fornecido.

Abra o Spark-Shell

O seguinte comando é usado para abrir o shell do Spark. Geralmente, o Spark é construído usando Scala. Portanto, um programa Spark é executado no ambiente Scala.

$ spark-shell

Se o shell do Spark abrir com êxito, você encontrará a seguinte saída. Observe a última linha da saída "Contexto do Spark disponível como sc" significa que o contêiner do Spark é criado automaticamente para o objeto de contexto do Spark com o nomesc. Antes de iniciar a primeira etapa de um programa, o objeto SparkContext deve ser criado.

Spark assembly has been built with Hive, including Datanucleus jars on classpath 
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties 
15/06/04 15:25:22 INFO SecurityManager: Changing view acls to: hadoop 
15/06/04 15:25:22 INFO SecurityManager: Changing modify acls to: hadoop 
15/06/04 15:25:22 INFO SecurityManager: SecurityManager: authentication disabled;
   ui acls disabled; users with view permissions: Set(hadoop); users with modify permissions: Set(hadoop) 
15/06/04 15:25:22 INFO HttpServer: Starting HTTP Server 
15/06/04 15:25:23 INFO Utils: Successfully started service 'HTTP class server' on port 43292. 
Welcome to 
      ____              __ 
     / __/__  ___ _____/ /__ 
    _\ \/ _ \/ _ `/ __/  '_/ 
   /___/ .__/\_,_/_/ /_/\_\   version 1.4.0 
      /_/  
		
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71) 
Type in expressions to have them evaluated. 
Spark context available as sc 
scala>

Crie um RDD

Primeiro, temos que ler o arquivo de entrada usando a API Spark-Scala e criar um RDD.

O comando a seguir é usado para ler um arquivo de um determinado local. Aqui, um novo RDD é criado com o nome de inputfile. O String que é fornecido como um argumento no método textFile (“”) é o caminho absoluto para o nome do arquivo de entrada. No entanto, se apenas o nome do arquivo for fornecido, isso significa que o arquivo de entrada está no local atual.

scala> val inputfile = sc.textFile("input.txt")

Executar transformação de contagem de palavras

Nosso objetivo é contar as palavras em um arquivo. Crie um mapa plano para dividir cada linha em palavras (flatMap(line ⇒ line.split(“ ”))

Em seguida, leia cada palavra como uma chave com um valor ‘1’ (<chave, valor> = <palavra, 1>) usando a função de mapa (map(word ⇒ (word, 1))

Por fim, reduza essas chaves adicionando valores de chaves semelhantes (reduceByKey(_+_))

O seguinte comando é usado para executar a lógica de contagem de palavras. Depois de executar isso, você não encontrará nenhuma saída porque esta não é uma ação, é uma transformação; apontar um novo RDD ou dizer à centelha o que fazer com os dados fornecidos)

scala> val counts = inputfile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_+_);

RDD atual

Ao trabalhar com o RDD, se desejar saber sobre o RDD atual, use o seguinte comando. Ele mostrará a descrição sobre o RDD atual e suas dependências para depuração.

scala> counts.toDebugString

Armazenando as Transformações

Você pode marcar um RDD para ser persistido usando os métodos persist () ou cache () nele. Na primeira vez em que for computado em uma ação, ele será mantido na memória dos nós. Use o seguinte comando para armazenar as transformações intermediárias na memória.

scala> counts.cache()

Aplicando a Ação

Aplicar uma ação, como armazenar todas as transformações, resulta em um arquivo de texto. O argumento String para o método saveAsTextFile (“”) é o caminho absoluto da pasta de saída. Tente o seguinte comando para salvar a saída em um arquivo de texto. No exemplo a seguir, a pasta 'output' está no local atual.

scala> counts.saveAsTextFile("output")

Verificando o resultado

Abra outro terminal para ir para o diretório inicial (onde o spark é executado no outro terminal). Use os seguintes comandos para verificar o diretório de saída.

[hadoop@localhost ~]$ cd output/ 
[hadoop@localhost output]$ ls -1 
 
part-00000 
part-00001 
_SUCCESS

O seguinte comando é usado para ver a saída de Part-00000 arquivos.

[hadoop@localhost output]$ cat part-00000

Resultado

(people,1) 
(are,2) 
(not,1) 
(as,8) 
(beautiful,2) 
(they, 7) 
(look,1)

O seguinte comando é usado para ver a saída de Part-00001 arquivos.

[hadoop@localhost output]$ cat part-00001

Resultado

(walk, 1) 
(or, 1) 
(talk, 1) 
(only, 1) 
(love, 1) 
(care, 1) 
(share, 1)

ONU Persiste o Armazenamento

Antes de persistir com o UN, se quiser ver o espaço de armazenamento usado para este aplicativo, use a seguinte URL em seu navegador.

http://localhost:4040

Você verá a tela a seguir, que mostra o espaço de armazenamento usado para o aplicativo, que está sendo executado no shell do Spark.

Se você deseja UN-persistir o espaço de armazenamento de um RDD específico, use o seguinte comando.

Scala> counts.unpersist()

Você verá a saída da seguinte forma -

15/06/27 00:57:33 INFO ShuffledRDD: Removing RDD 9 from persistence list 
15/06/27 00:57:33 INFO BlockManager: Removing RDD 9 
15/06/27 00:57:33 INFO BlockManager: Removing block rdd_9_1 
15/06/27 00:57:33 INFO MemoryStore: Block rdd_9_1 of size 480 dropped from memory (free 280061810) 
15/06/27 00:57:33 INFO BlockManager: Removing block rdd_9_0 
15/06/27 00:57:33 INFO MemoryStore: Block rdd_9_0 of size 296 dropped from memory (free 280062106) 
res7: cou.type = ShuffledRDD[9] at reduceByKey at <console>:14

Para verificar o espaço de armazenamento no navegador, use a seguinte URL.

http://localhost:4040/

Você verá a seguinte tela. Ele mostra o espaço de armazenamento usado para o aplicativo, que está sendo executado no shell do Spark.

O aplicativo Spark, usando spark-submit, é um comando shell usado para implantar o aplicativo Spark em um cluster. Ele usa todos os respectivos gerenciadores de cluster por meio de uma interface uniforme. Portanto, você não precisa configurar seu aplicativo para cada um.

Exemplo

Tomemos o mesmo exemplo de contagem de palavras que usamos antes, usando comandos shell. Aqui, consideramos o mesmo exemplo de um aplicativo Spark.

Amostra de entrada

O texto a seguir são os dados de entrada e o arquivo nomeado é in.txt.

people are not as beautiful as they look, 
as they walk or as they talk. 
they are only as beautiful  as they love, 
as they care as they share.

Veja o seguinte programa -

SparkWordCount.scala

import org.apache.spark.SparkContext 
import org.apache.spark.SparkContext._ 
import org.apache.spark._  

object SparkWordCount { 
   def main(args: Array[String]) { 

      val sc = new SparkContext( "local", "Word Count", "/usr/local/spark", Nil, Map(), Map()) 
		
      /* local = master URL; Word Count = application name; */  
      /* /usr/local/spark = Spark Home; Nil = jars; Map = environment */ 
      /* Map = variables to work nodes */ 
      /*creating an inputRDD to read text file (in.txt) through Spark context*/ 
      val input = sc.textFile("in.txt") 
      /* Transform the inputRDD into countRDD */ 
		
      val count = input.flatMap(line ⇒ line.split(" ")) 
      .map(word ⇒ (word, 1)) 
      .reduceByKey(_ + _) 
       
      /* saveAsTextFile method is an action that effects on the RDD */  
      count.saveAsTextFile("outfile") 
      System.out.println("OK"); 
   } 
}

Salve o programa acima em um arquivo chamado SparkWordCount.scala e coloque-o em um diretório definido pelo usuário chamado spark-application.

Note - Ao transformar o inputRDD em countRDD, estamos usando flatMap () para tokenizar as linhas (do arquivo de texto) em palavras, o método map () para contar a frequência da palavra e o método reduceByKey () para contar cada repetição de palavra.

Use as etapas a seguir para enviar este aplicativo. Execute todas as etapas nospark-application diretório através do terminal.

Etapa 1: Baixe o Spark Ja

Spark core jar é necessário para a compilação, portanto, baixe spark-core_2.10-1.3.0.jar do seguinte link Spark core jar e mova o arquivo jar do diretório de download paraspark-application diretório.

Etapa 2: compilar o programa

Compile o programa acima usando o comando fornecido abaixo. Este comando deve ser executado a partir do diretório do aplicativo spark. Aqui,/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar é um jar de suporte do Hadoop obtido da biblioteca Spark.

$ scalac -classpath "spark-core_2.10-1.3.0.jar:/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar" SparkPi.scala

Etapa 3: Criar um JAR

Crie um arquivo jar do aplicativo spark usando o seguinte comando. Aqui,wordcount é o nome do arquivo para o arquivo jar.

jar -cvf wordcount.jar SparkWordCount*.class spark-core_2.10-1.3.0.jar/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar

Etapa 4: enviar o aplicativo Spark

Envie o aplicativo Spark usando o seguinte comando -

spark-submit --class SparkWordCount --master local wordcount.jar

Se for executado com sucesso, você encontrará a saída fornecida abaixo. oOKdeixar a seguinte saída é para identificação do usuário e essa é a última linha do programa. Se você ler cuidadosamente a saída a seguir, encontrará coisas diferentes, como -

serviço iniciado com sucesso 'sparkDriver' na porta 42954
MemoryStore começou com capacidade de 267,3 MB
SparkUI iniciado em http://192.168.1.217:4040
Arquivo JAR adicionado: /home/hadoop/piapplication/count.jar
ResultStage 1 (saveAsTextFile em SparkPi.scala: 11) terminou em 0,566 s
Interrompeu a interface do usuário da Web do Spark em http://192.168.1.217:4040
MemoryStore apagado

15/07/08 13:56:04 INFO Slf4jLogger: Slf4jLogger started 
15/07/08 13:56:04 INFO Utils: Successfully started service 'sparkDriver' on port 42954. 
15/07/08 13:56:04 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://[email protected]:42954] 
15/07/08 13:56:04 INFO MemoryStore: MemoryStore started with capacity 267.3 MB 
15/07/08 13:56:05 INFO HttpServer: Starting HTTP Server 
15/07/08 13:56:05 INFO Utils: Successfully started service 'HTTP file server' on port 56707. 
15/07/08 13:56:06 INFO SparkUI: Started SparkUI at http://192.168.1.217:4040 
15/07/08 13:56:07 INFO SparkContext: Added JAR file:/home/hadoop/piapplication/count.jar at http://192.168.1.217:56707/jars/count.jar with timestamp 1436343967029 
15/07/08 13:56:11 INFO Executor: Adding file:/tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af/userFiles-df4f4c20-a368-4cdd-a2a7-39ed45eb30cf/count.jar to class loader 
15/07/08 13:56:11 INFO HadoopRDD: Input split: file:/home/hadoop/piapplication/in.txt:0+54 
15/07/08 13:56:12 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2001 bytes result sent to driver 
 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11), which is now runnable 
15/07/08 13:56:12 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11) 
15/07/08 13:56:13 INFO DAGScheduler: ResultStage 1 (saveAsTextFile at SparkPi.scala:11) finished in 0.566 s 
15/07/08 13:56:13 INFO DAGScheduler: Job 0 finished: saveAsTextFile at SparkPi.scala:11, took 2.892996 s
OK 
15/07/08 13:56:13 INFO SparkContext: Invoking stop() from shutdown hook 
15/07/08 13:56:13 INFO SparkUI: Stopped Spark web UI at http://192.168.1.217:4040 
15/07/08 13:56:13 INFO DAGScheduler: Stopping DAGScheduler 
15/07/08 13:56:14 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped! 
15/07/08 13:56:14 INFO Utils: path = /tmp/spark-45a07b83-42ed-42b3-b2c2823d8d99c5af/blockmgr-ccdda9e3-24f6-491b-b509-3d15a9e05818, already present as root for deletion. 
15/07/08 13:56:14 INFO MemoryStore: MemoryStore cleared 
15/07/08 13:56:14 INFO BlockManager: BlockManager stopped 
15/07/08 13:56:14 INFO BlockManagerMaster: BlockManagerMaster stopped 
15/07/08 13:56:14 INFO SparkContext: Successfully stopped SparkContext 
15/07/08 13:56:14 INFO Utils: Shutdown hook called 
15/07/08 13:56:14 INFO Utils: Deleting directory /tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af 
15/07/08 13:56:14 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!

Etapa 5: verificar a saída

Após a execução bem-sucedida do programa, você encontrará o diretório chamado outfile no diretório do aplicativo spark.

Os comandos a seguir são usados para abrir e verificar a lista de arquivos no diretório outfile.

$ cd outfile 
$ ls 
Part-00000 part-00001 _SUCCESS

Os comandos para verificar a saída em part-00000 arquivo são -

$ cat part-00000 
(people,1) 
(are,2) 
(not,1) 
(as,8) 
(beautiful,2) 
(they, 7) 
(look,1)

Os comandos para verificar a saída no arquivo parte-00001 são -

$ cat part-00001 
(walk, 1) 
(or, 1) 
(talk, 1) 
(only, 1) 
(love, 1) 
(care, 1) 
(share, 1)

Vá até a seção a seguir para saber mais sobre o comando 'spark-submit'.

Sintaxe de envio do Spark

spark-submit [options] <app jar | python file> [app arguments]

Opções

A tabela abaixo descreve uma lista de options -

S.No	Opção	Descrição
1	--mestre	spark: // host: port, mesos: // host: port, yarn ou local.
2	--deploy-mode	Se deve iniciar o programa de driver localmente ("cliente") ou em uma das máquinas de trabalho dentro do cluster ("cluster") (Padrão: cliente).
3	--classe	A classe principal do seu aplicativo (para aplicativos Java / Scala).
4	--nome	Um nome de seu aplicativo.
5	--jars	Lista separada por vírgulas de jars locais para incluir nos caminhos de classe do driver e do executor.
6	- pacotes	Lista separada por vírgulas de coordenadas maven de jars para incluir nos caminhos de classe do driver e do executor.
7	--repositórios	Lista separada por vírgulas de repositórios remotos adicionais para pesquisar as coordenadas maven fornecidas com --packages.
8	--py-files	Lista separada por vírgulas de arquivos .zip, .egg ou .py para colocar no PYTHON PATH para aplicativos Python.
9	--arquivos	Lista separada por vírgulas de arquivos a serem colocados no diretório de trabalho de cada executor.
10	--conf (prop = val)	Propriedade de configuração arbitrária do Spark.
11	--properties-file	Caminho para um arquivo do qual carregar propriedades extras. Se não for especificado, ele procurará conf / spark-defaults.
12	--driver-memory	Memória para driver (por exemplo, 1000M, 2G) (Padrão: 512M).
13	--driver-java-options	Opções extras de Java para passar para o driver.
14	--driver-library-path	Entradas de caminho de biblioteca extras para passar para o driver.
15	--driver-class-path	Entradas de caminho de classe extras para passar ao motorista. Observe que os jars adicionados com --jars são incluídos automaticamente no caminho de classe.
16	--executor-memory	Memória por executor (por exemplo, 1000M, 2G) (Padrão: 1G).
17	--proxy-user	Usuário para personificar ao enviar o aplicativo.
18	--help, -h	Mostre esta mensagem de ajuda e saia.
19	--verbose, -v	Imprime saída de depuração adicional.
20	--versão	Imprima a versão do Spark atual.
21	--driver-cores NUM	Cores para driver (Padrão: 1).
22	--supervisionar	Se fornecido, reinicia o driver em caso de falha.
23	--matar	Se fornecido, mata o driver especificado.
24	--status	Se fornecido, solicita o status do driver especificado.
25	--total-executor-cores	Total de núcleos para todos os executores.
26	--executor-cores	Número de núcleos por executor. (Padrão: 1 no modo YARN ou todos os núcleos disponíveis no trabalhador no modo autônomo).

Spark contém dois tipos diferentes de variáveis compartilhadas - uma é broadcast variables e o segundo é accumulators.

Broadcast variables - usado para distribuir grandes valores com eficiência.
Accumulators - usado para agregar as informações de uma coleção particular.

Variáveis de transmissão

Variáveis de transmissão permitem que o programador mantenha uma variável somente leitura em cache em cada máquina, em vez de enviar uma cópia dela com as tarefas. Eles podem ser usados, por exemplo, para dar a cada nó, uma cópia de um grande conjunto de dados de entrada, de maneira eficiente. O Spark também tenta distribuir variáveis de transmissão usando algoritmos de transmissão eficientes para reduzir o custo de comunicação.

As ações do Spark são executadas por meio de um conjunto de estágios, separados por operações "shuffle" distribuídas. O Spark transmite automaticamente os dados comuns necessários às tarefas em cada estágio.

Os dados transmitidos dessa maneira são armazenados em cache na forma serializada e são desserializados antes de executar cada tarefa. Isso significa que criar variáveis de transmissão explicitamente só é útil quando as tarefas em vários estágios precisam dos mesmos dados ou quando é importante armazenar os dados em cache na forma desserializada.

Variáveis de transmissão são criadas a partir de uma variável v chamando SparkContext.broadcast(v). A variável de transmissão é um wrapper ao redorv, e seu valor pode ser acessado chamando o valuemétodo. O código fornecido abaixo mostra isso -

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

Output -

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

Depois que a variável de transmissão é criada, ela deve ser usada em vez do valor v em qualquer função executada no cluster, de modo que vnão é enviado para os nós mais de uma vez. Além disso, o objetov não deve ser modificado após sua transmissão, para garantir que todos os nós obtenham o mesmo valor da variável de transmissão.

Acumuladores

Acumuladores são variáveis que só são “adicionadas” por meio de uma operação associativa e, portanto, podem ser eficientemente suportados em paralelo. Eles podem ser usados para implementar contadores (como no MapReduce) ou somas. O Spark oferece suporte nativo para acumuladores de tipos numéricos, e os programadores podem adicionar suporte para novos tipos. Se os acumuladores forem criados com um nome, eles serão exibidos emSpark’s UI. Isso pode ser útil para entender o progresso dos estágios em execução (NOTA - ainda não é compatível com Python).

Um acumulador é criado a partir de um valor inicial v chamando SparkContext.accumulator(v). As tarefas em execução no cluster podem ser adicionadas a ele usando oaddou o operador + = (em Scala e Python). No entanto, eles não podem ler seu valor. Apenas o programa driver pode ler o valor do acumulador, usando seuvalue método.

O código fornecido abaixo mostra um acumulador sendo usado para adicionar os elementos de uma matriz -

scala> val accum = sc.accumulator(0) 
 
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

Se você quiser ver a saída do código acima, use o seguinte comando -

scala> accum.value

Resultado

res2: Int = 10

Operações numéricas de RDD

O Spark permite que você faça diferentes operações em dados numéricos, usando um dos métodos de API predefinidos. As operações numéricas do Spark são implementadas com um algoritmo de streaming que permite construir o modelo, um elemento de cada vez.

Essas operações são calculadas e retornadas como um StatusCounter Objeto chamando status() método.

A seguir está uma lista de métodos numéricos disponíveis em StatusCounter.

S.No	Métodos e Significado
1	count() Número de elementos do RDD.
2	Mean() Média dos elementos do RDD.
3	Sum() Valor total dos elementos do RDD.
4	Max() Valor máximo entre todos os elementos do RDD.
5	Min() Valor mínimo entre todos os elementos do RDD.
6	Variance() Variância dos elementos.
7	Stdev() Desvio padrão.