A necessidade
Ah, as migrações de banco de dados. Depois de migrar para o Couchbase, em que a representação de dados do documento JSON é muito mais flexível, você não precisará mais distorcer seu processo de desenvolvimento com tanta frequência por meio do pipeline de solicitações de alteração, seguido pelo tempo de inatividade, possíveis erros e o desagrado geral quando vinculado a linhas e colunas.
Dito isso, o fato de estar no Couchbase não significa que você estará sempre livre da necessidade de fazer a transformação de dados com dados no Couchbase. Isso será apenas muito menos comum.
Esse é, de fato, um lugar interessante para o Apache Spark!
Um aparte...
Com o conector Spark e o Couchbase 4.0, você tem não apenas uma, mas quatro interfaces que são relevantes para quem usa o Spark. São elas: a interface K-V, a interface de streaming Database Change Protocol (também conhecida como DCP), a interface de consulta N1QL via Spark SQL e a interface View Query.
Eles podem ser combinados com uma série de fontes de dados diferentes do ecossistema do Spark para reunir e manipular dados de várias maneiras. Por exemplo, você pode querer transmitir os dados do Couchbase por meio do DCP, misturá-los com uma fonte de dados do HDFS e colocar os resultados de destino de volta em um bucket diferente do Couchbase.
A solução...
Tomando um caso simples, como podemos usar o Spark para escrever algum código para transformar com eficiência um conjunto de dados dentro do Couchbase?
Imagine o cenário em que você adquiriu um novo conjunto de dados sobre jogadores em formato JSON. Todos eles estarão jogando seu novo jogo FizzBuzz em breve, e os perfis foram enviados por um parceiro. Todos os perfis recebidos são parecidos com este:
|
1 2 3 4 5 6 |
{ "nome próprio": "Joel", "sobrenome": "Smith", "e-mail": "joelsmith@g00glemail.com", "token de direito": 78238743 } |
O problema é que os perfis do FizzBuzz são todos parecidos com este:
|
1 2 3 4 5 6 |
{ "fname": "Matt", "nome": "Ingenthron", "e-mail": "fosco@couchbase.com", "currentscore": 1000000 } |
Normalmente, se você tivesse outra forma para os dados, adicionaria um pouco de lógica para o mapeamento no momento da leitura e da gravação. No entanto, essa transição específica é um processo único e vem com um detalhe adicional. Esse "entitlementtoken" precisa ser consultado em um backup do banco de dados MySQL que você também possui. Você não quer ter que provisionar ou manter uma grande implementação do MySQL para lidar com o tráfego do dia do lançamento, portanto, uma transformação única antes do lançamento é melhor.
O ideal seria transmitir os dados, encontrar aqueles com a "forma" que desejamos e transformá-los com o Spark com base em uma consulta SQL.
Primeiro, precisamos configurar nossa conexão e transmitir os dados, procurando a forma do JSON já importado. Isso usará a interface DCP do Couchbase para transmitir os dados.
|
1 2 3 4 5 6 |
val ssc = novo Contexto de fluxo contínuo(sc, Segundos(5)) ssc.couchbaseStream("transformador") .filtro(_.isInstanceOf[Mutação]) .mapa(m => (novo Cordas(m.asInstanceOf[Mutação].chave), novo Cordas(m.asInstanceOf[Mutação].conteúdo))) |
Uma limitação atual é que o DStream nunca para, mas podemos simplesmente monitorar quando não vemos mais dados sendo transformados como uma solução alternativa para esse caso simples.
Em seguida, por item, precisamos aplicar uma transformação com base nessa pesquisa do MySQL. Para fazer isso, precisaremos carregar os dados do MySQL. Supondo que a tabela do MySQL tenha a seguinte aparência:
|
1 2 3 4 5 6 7 8 9 10 |
mysql> descrever perfis; +------------------+-------------+------+-----+---------+-------+ | Campo | Tipo | Nulo | Chave | Padrão | Extra | +------------------+-------------+------+-----+---------+-------+ | nome próprio | varchar(20) | SIM | | NULL | | | sobrenome | varchar(20) | SIM | | NULL | | | e-mail | varchar(20) | SIM | | NULL | | | token de direito | int(11) | SIM | | NULL | | +------------------+-------------+------+-----+---------+-------+ 4 linhas em definir (0.00 sec) |
Queremos carregar os dados do MySQL como um DataFrame. Como o StreamingContext nos fornece RDDs para unir, nós o converteremos em um conjunto de RDDs para uma união posterior dentro do fluxo. O Spark 1.6 pode tornar isso mais fácil. Essa conversão tem a seguinte aparência (extraída para uma função para facilitar a leitura):
|
1 2 3 4 5 6 |
/** Retorna um RDD com base no endereço de e-mail extraído do documento */ def CreateMappableRdd(s: (Cordas, Cordas)): (Cordas, JsonDocument) = { val return_doc = JsonDocument.criar(s._1, JsonObject.fromJson(s._2)) (return_doc.conteúdo().getString("email"), return_doc) } |
Também precisamos adicionar o novo token de direito (também extraído):
|
1 2 3 4 5 6 7 8 |
/** Retorna um JsonDocument enriquecido com o token de direito */ def mergeIntoDoc(t: (Cordas, (JsonDocument, Inteiro))): JsonDocument = { val jsonToEnrich = t._2._1.conteúdo() val entitlementFromJoin = t._2._2 jsonToEnrich.colocar("entitlementtoken", entitlementFromJoin) t._2._1 } |
No final, temos uma boa descrição fluente da nossa transformação, modificando os RDDs em voo que precisam de alterações. Por fim, isso grava os dados transformados de volta no Couchbase, substituindo os itens usando a interface K-V.
|
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 |
// carregar o DataFrame de todos os usuários do MySQL. // Observação: anexar .cache() pode fazer sentido aqui (ou não), dependendo da quantidade de dados. val direitos = mysqlReader.carregar() /* carregando isso: +---------+-----------+-----------------+----------------+ |givenname| surname| email|entitlementtoken| +---------+-----------+-----------------+----------------+ | Matt| Ingenthron| matt@email.com| 11211| | Michael|Nitschinger|michael@email.com| 11210| +---------+-----------+-----------------+----------------+ */ val entitlementsSansSchema = direitos.rdd.mapa[(Cordas, Inteiro)](f => (f.getAs[Cordas]("email"), f.getAs[Inteiro]("entitlementtoken"))) val ssc = novo Contexto de fluxo contínuo(sc, Segundos(5)) ssc.couchbaseStream("transformador") .filtro(_.isInstanceOf[Mutação]) .mapa(m => (novo Cordas(m.asInstanceOf[Mutação].chave), novo Cordas(m.asInstanceOf[Mutação].conteúdo))) .mapa(s => CreateMappableRdd(s)) .filtro(_._2.conteúdo().obter("entitlementtoken").eq(nulo)) .foreachRDD(rdd => { rdd .unir-se(entitlementsSansSchema) .mapa(mergeIntoDoc) //.foreach(println) // um bom lugar para ver o efeito .saveToCouchbase("transformador") }) ssc.iniciar() ssc.awaitTermination() |
O O exemplo completo está no couchbase-spark-samples repositório.
A beleza desse exemplo é que ele é fácil de entender o que está acontecendo e bastante trivial para ser dimensionado. É provável que sua própria transformação seja mais complexa, mas esse exemplo deve lhe dar uma noção do que é possível e algo a ser desenvolvido.
Sempre há espaço para melhorias.
Um problema é que o MySQL pode ser maior do que o que eu quero carregar na memória. O Spark leva isso em conta ao oferecer uma maneira de dividir os DataFrames. Eu não precisava disso aqui e queria que a amostra fosse legível. Outra coisa que pode ajudar nisso é a capacidade de fazer referência a um SparkContext em um StreamingContext existente. O Spark não permite isso no momento por boas razões, mas eu diria que esse caso de uso simples de fazer uma pesquisa de registro único de dentro do fluxo faz sentido.
No Conector Couchbase, no momento, a interface DCP é classificada como volátil e deve ser considerada experimental. Além disso, o exemplo acima é muito rápido, mas precisa de ajuda para ser dimensionado. Uma atualização futura do meu colega Sergey Avseyev permitirá dividir os fluxos de DCP entre os trabalhadores do Spark para paralelizar essa transformação.
Para concluir
O Spark é uma nova e excelente ferramenta para esse tipo de transformação. As mesmas técnicas podem certamente ser aplicadas à migração para o Couchbase a partir de uma fonte de dados diferente, como um banco de dados relacional. A técnica pode até ser expandida com o aprendizado de máquina do Spark para criar um modelo em torno do fluxo de dados do Couchbase para antecipar resultados.
