Transferência de dados do Cassandra para o Couchbase usando o Spark

Shivansh Srivastava é um desenvolvedor poliglota e um aficionado por Scala e Spark Ele gosta de contribuir para projetos de código aberto e contribuiu para vários projetos, como Apache Iota, Apache Spark e Apache Carbondata, Couchbase-Spark-ConnectorAkka, Alpakka e muitos outros projetos. Ele tem um grande interesse em tecnologias futuras, como IoT, DeepLearning etc. CAtualmente, ele ocupa o cargo de engenheiro de software sênior na Chirpanywhere Inc., uma startup baseada em IoT, onde gerencia muitas coisas, desde a programação de hardware até o projeto de toda a solução e sua implementação. 

Há muitos Bancos de dados NoSQL no mercado, como Cassandra, MongoDB, Couchbase e outros, e cada um deles tem prós e contras.

Tipos de bancos de dados NoSQL

Há principalmente quatro tipos de bancos de dados NoSQL, a saber:

1. Orientado para colunas

2. Armazenamento de valores-chave

3. Orientado a documentos

4. Gráfico

Os bancos de dados que oferecem suporte a mais de um formato são chamados de "multimodelos", como o Couchbase, que oferece suporte a bancos de dados orientados a documentos e a valores-chave.

Às vezes, escolhemos o banco de dados errado para o nosso aplicativo e percebemos essa dura verdade em um estágio posterior.

E depois? O que devemos fazer?

Esse é o caso da nossa experiência, em que estávamos usando o Cassandra como nosso banco de dados e, mais tarde, descobrimos que ele não estava atendendo a todas as nossas necessidades. Precisávamos encontrar um novo banco de dados e descobrimos que o Couchbase era a opção certa.

A principal dificuldade foi descobrir como deveríamos transferir nossos dados de Cassandra para Couchbaseporque esse plug-in não estava disponível.

Nesta postagem do blog, descreverei o código que escrevi para transferir dados do Cassandra para o Couchbase usando o Spark.

Todo o código está disponível aqui: cassandra-couchbase-transfer-plugin

Explicação do código

Aqui, estou lendo dados do Cassandra e gravando-os de volta no Couchbase. Esse código simples resolve nosso problema.

As etapas envolvidas são:

1. Leitura da configuração:
   val configuração = ConfigFactory.carregar()
   //Configuração doouchbase
   val bucketName = configuração.getString("couchbase.bucketName")
   val couchbaseHost = configuração.getString("couchbase.host")
   //Configuração do Cassandra
   val keyspaceName = configuração.getString("cassandra.keyspaceName")
   val tableName = configuração.getString("cassandra.tableName")
   val idFeild = configuração.getString("cassandra.idFeild")
   val cassandraHost = configuração.getString("cassandra.host")
   val cassandraPort = configuração.getInt("cassandra.port")

2. Definição da configuração do Spark e criação da sessão do Spark:
   val conf = novo SparkConf()
     .setAppName(s "CouchbaseCassandraTransferPlugin")
    .setMaster("local[*]")
    .set(s "com.couchbase.bucket.$bucketName“, “”)
    .set("com.couchbase.nodes", couchbaseHost)
    .set("spark.cassandra.connection.host", cassandraHost)
    .set("spark.cassandra.connection.port", cassandraPort.toString)
   val faísca = SparkSession.construtor().config(conf).getOrCreate()
   val sc = faísca.sparkContext

3. Leitura de dados do Cassandra:
     val cassandraRDD = faísca.leia
     .format("org.apache.spark.sql.cassandra")
     .opções(Mapa("tabela" -> tableName, "espaço-chave" -> keyspaceName))
     .load()

4. Verificando o campo id:
   O campo id está sendo verificado para ver se ele existe. Em seguida, use-o também como id no Couchbase ou gere um id aleatório e atribua-o ao documento.
   importação org.apache.spark.sql.functions._
   val uuidUDF = udf(CouchbaseHelper.getUUID _)
   val rddToBeWritten = se (cassandraRDD.columns.contains(idFeild)) {
     cassandraRDD.withColumn("META_ID", cassandraRDD(idFeild))
   } mais {
     cassandraRDD.withColumn("META_ID", uuidUDF())
   }
   
   Em um arquivo diferente:
   objeto CouchbaseHelper {
   def getUUID: Cordas = UUID.UUUID aleatório().toString
   }

5. Gravação no Couchbase:
   rddToBeWritten.write.couchbase()

Você pode executar esse código diretamente para transferir dados do Cassandra para o Couchbase - Tudo o que você precisa fazer é uma configuração.

Configurações

Todas as configurações podem ser feitas definindo o parâmetro variáveis de ambiente.

1. Configuração do Couchbase:

2. Configuração do Cassandra:

Código em ação

Lado Cassandra:

Esta é a aparência dos dados no lado do Cassandra.

Lado do Couchbase:

Caso 1: Quando o id existe e pode ser usado como id do documento do Couchbase.

Caso 2: Quando o nome do ID não existe e precisamos atribuir um ID aleatório aos documentos.

Como executar o plug-in de transferência Cassandra-Couchbase

Etapas para executar o código:

1. Baixe o código do repositório.

2. Configure as variáveis de ambiente de acordo com a configuração.

3. Execute o projeto usando execução do sbt

Deseja escrever sobre o Couchbase e participar de nosso programa de redação da comunidade? Saiba mais!