Migrar dados dinâmicos do Apache Cassandra para o Azure Cosmos DB for Apache Cassandra usando o proxy de gravação dupla e o Apache Spark
O API para Cassandra no Azure Cosmos DB se tornou uma ótima opção para cargas de trabalho corporativas em execução no Apache Cassandra por vários motivos, como:
Sem sobrecarga de gerenciamento e monitoramento: Ele elimina a sobrecarga de gerenciar e monitorar uma infinidade de configurações em arquivos de sistema operacional, de JVM e arquivos YAML e suas interações.
Economia de custos significativa: Você pode economizar custos com Azure Cosmos DB, o que inclui o custo da VM, a largura de banda e todas as licenças aplicáveis. Além disso, você não precisa gerenciar os custos de data centers, servidores, armazenamento de SSD, rede e eletricidade.
Capacidade de usar o código e as ferramentas existentes: o Azure Cosmos DB fornece compatibilidade no nível de protocolo com as ferramentas e os SDKs existentes do Cassandra. Essa compatibilidade garante que seja possível usar a base de código existente com o Azure Cosmos DB for Apache Cassandra com alterações básicas.
O Azure Cosmos DB não dá suporte ao protocolo nativo de bisbilhotice Apache Para replicação. Portanto, onde o tempo de inatividade zero é um requisito para migração, uma abordagem diferente é necessária. Este tutorial descreve como migrar dados dinâmicos para o Azure Cosmos DB for Apache Cassandra de um cluster Apache Cassandra nativo usando um proxy de gravação dupla e o Apache Spark.
A imagem a seguir ilustra o padrão. O proxy de gravação dupla é usado para capturar alterações ao vivo, enquanto os dados históricos são copiados em massa usando o Apache Spark. O proxy pode aceitar conexões do código do aplicativo com pouca ou nenhuma alteração de configuração. Ele roteia todas as solicitações para o banco de dados de origem e roteia de maneira assíncrona gravações para a API para Cassandra enquanto a cópia em massa está acontecendo.
Pré-requisitos
Provisionar uma conta do Azure Cosmos DB for Apache Cassandra.
Examinar os conceitos básicos da conexão com um Azure Cosmos DB for Apache Cassandra.
Examinar os recursos com suporte no Azure Cosmos DB for Apache Cassandra para garantir a compatibilidade.
Verifique se você tem conectividade de rede entre o cluster de origem e o ponto de extremidade da API para Cassandra de destino.
Garanta que você já migrou o esquema de keyspace/tabela do banco de dados de origem do Cassandra para a conta de API para Cassandra de destino.
Importante
Se você tiver um requisito para preservar Apache Cassandra durante a migração
writetimeos seguintes sinalizadores devem ser definidos ao criar tabelas:with cosmosdb_cell_level_timestamp=true and cosmosdb_cell_level_timestamp_tombstones=true and cosmosdb_cell_level_timetolive=truePor exemplo:
CREATE KEYSPACE IF NOT EXISTS migrationkeyspace WITH REPLICATION= {'class': 'org.apache.> cassandra.locator.SimpleStrategy', 'replication_factor' : '1'};CREATE TABLE IF NOT EXISTS migrationkeyspace.users ( name text, userID int, address text, phone int, PRIMARY KEY ((name), userID)) with cosmosdb_cell_level_timestamp=true and > cosmosdb_cell_level_timestamp_tombstones=true and cosmosdb_cell_level_timetolive=true;
Provisionar um cluster Spark
Recomendamos o Azure Databricks. Use um tempo de execução compatível com o Spark 3.0 ou superior.
Importante
Você precisa garantir que sua conta do Azure Databricks tenha conectividade de rede com seu cluster de origem Apache Cassandra. Isso pode exigir injeção de VNet. Consulte o artigo aqui para obter mais informações.
Adicionar dependências do Spark
Você precisa adicionar a biblioteca do Conector do Apache Spark Cassandra ao cluster para se conectar aos pontos de extremidade nativos e do Cassandra do Azure Cosmos DB. No cluster, selecione Bibliotecas>Instalar novo>Maven e, em seguida, adicione com.datastax.spark:spark-cassandra-connector-assembly_2.12:3.0.0 nas coordenadas do Maven.
Importante
Se você tiver um requisito para preservar o writetime do Apache Cassandra para cada linha durante a migração, recomendamos usar esta amostra. O jar de dependência neste exemplo também contém o conector do Spark, portanto, você deve instalá-lo em vez do assembly do conector acima. Este exemplo também será útil, se você quiser executar uma validação de comparação de linha entre a origem e o destino, após a conclusão da carga de dados histórica. Confira as seções "executar a carga de dados históricos" e "validar a origem e o destino" abaixo para obter mais detalhes.
Selecione Instalar e reinicie o cluster quando a instalação for concluída.
Observação
Verifique se reinicia o cluster do Azure Databricks após a instalação da biblioteca do Cassandra Connector.
Instalar o proxy de gravação dupla
Para obter um desempenho ideal durante gravações duplas, é recomendável instalar o proxy em todos os nós do cluster do Cassandra de origem.
#assuming you do not have git already installed
sudo apt-get install git
#assuming you do not have maven already installed
sudo apt install maven
#clone repo for dual-write proxy
git clone https://github.com/Azure-Samples/cassandra-proxy.git
#change directory
cd cassandra-proxy
#compile the proxy
mvn package
Iniciar o proxy de gravação dupla
Recomendamos instalar o proxy em todos os nós do cluster do Cassandra de origem. No mínimo, execute o comando a seguir para iniciar o proxy em cada nó. Substitua <target-server> por um endereço IP ou de servidor de um dos nós no cluster de destino. Substitua <path to JKS file> pelo caminho para um arquivo .jks local e <keystore password> pela senha correspondente.
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar localhost <target-server> --proxy-jks-file <path to JKS file> --proxy-jks-password <keystore password>
Ao iniciar o proxy dessa maneira, pressupõe-se que o seguinte é verdadeiro:
- O nome de usuário e a senha dos pontos de extremidade de origem e de destino são iguais.
- Os pontos de extremidade de origem e destino implementam protocolo SSL.
Se os pontos de extremidade de origem e destino não atenderem a esses critérios, continue lendo para obter mais opções de configuração.
Configurar SSL
Para o SSL, você pode implementar um repositório de chaves existente (por exemplo, aquele usado pelo cluster de origem) ou criar um certificado autoassinado usando o keytool:
keytool -genkey -keyalg RSA -alias selfsigned -keystore keystore.jks -storepass password -validity 360 -keysize 2048
Você também poderá desabilitar o SSL para pontos de extremidade de origem ou de destino caso eles não implementem o SSL. Use os sinalizadores --disable-source-tls ou --disable-target-tls:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar localhost <target-server> --source-port 9042 --target-port 10350 --proxy-jks-file <path to JKS file> --proxy-jks-password <keystore password> --target-username <username> --target-password <password> --disable-source-tls true --disable-target-tls true
Observação
Ao criar conexões SSL com o banco de dados por meio do proxy, verifique se o aplicativo cliente está usando o repositório de chaves e a senha iguais aos usados para o proxy de gravação dupla.
Configurar as credenciais e a porta
Por padrão, as credenciais de origem serão transmitidas a partir do seu aplicativo cliente. O proxy usará as credenciais para fazer conexões com os clusters de origem e de destino. Como mencionado antes, esse processo pressupõe que as credenciais de origem e de destino são as mesmas. Será necessário especificar um nome de usuário e senha diferentes para o ponto de extremidade da API para Cassandra de destino separadamente ao iniciar o proxy:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar localhost <target-server> --proxy-jks-file <path to JKS file> --proxy-jks-password <keystore password> --target-username <username> --target-password <password>
Quando não forem especificadas, as portas de origem e de destino padrão serão 9042. Nesse caso, a API para Cassandra é executada na porta 10350, portanto, você precisa usar --source-port ou --target-port para especificar números de porta:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar localhost <target-server> --source-port 9042 --target-port 10350 --proxy-jks-file <path to JKS file> --proxy-jks-password <keystore password> --target-username <username> --target-password <password>
Implantar o proxy remotamente
Pode haver circunstâncias em que você não deseje instalar o proxy nos nós de cluster em si e prefira instalá-lo em um computador separado. Nesse cenário, você precisa especificar o endereço IP de <source-server>:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar <source-server> <destination-server>
Aviso
Instalar e executar o proxy remotamente em uma máquina separada (em vez de executa-lo em todos os nós no cluster do Apache Cassandra de origem) afetará o desempenho durante a migração ao vivo. Embora funcione na prática, o driver do cliente não poderá abrir conexões com todos os nós no cluster e dependerá do nó de coordenação único (em que o proxy está instalado) para fazer conexões.
Permitir alterações de código de aplicativo zero
Por padrão, o proxy escuta na porta 29042. O código do aplicativo precisa ser alterado para apontar para essa porta. No entanto, você pode alterar a porta de escuta do proxy. Você pode fazer isso se quiser eliminar as alterações de código no nível do aplicativo da seguinte maneira:
- Fazendo com que o servidor Cassandra de origem seja executado em uma porta diferente.
- Fazendo com que o proxy seja executado na porta Cassandra padrão 9042.
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar source-server destination-server --proxy-port 9042
Observação
Na instalação do proxy em nós de cluster, não é preciso reiniciar os nós. No entanto, caso você tenha muitos clientes de aplicativo e prefira que o proxy execute na porta padrão do Cassandra 9042 para eliminar as alterações de código no nível do aplicativo, precisará alterar a porta padrão do Apache Cassandra. Em seguida, você precisará reiniciar os nós no cluster e configurar a porta de origem como a nova porta que você definiu para o cluster Cassandra de origem.
No exemplo a seguir, alteramos o cluster Cassandra de origem para execução na porta 3074 e iniciamos o cluster na porta 9042:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar source-server destination-server --proxy-port 9042 --source-port 3074
Forçar protocolos
O proxy tem funcionalidade para forçar protocolos, as quais podem ser necessárias se o ponto de extremidade de origem for mais avançado do que o de destino ou ele se não tiver suporte. Nesse caso, você pode especificar que --protocol-version e --cql-version forcem o protocolo a obedecer ao destino:
java -jar target/cassandra-proxy-1.0-SNAPSHOT-fat.jar source-server destination-server --protocol-version 4 --cql-version 3.11
Quando o proxy de gravação dupla estiver em execução, você precisará alterar a porta no cliente do aplicativo e reiniciar. Também poderá alterar a porta Cassandra e reiniciar o cluster se você tiver escolhido essa abordagem. O proxy começará a encaminhar gravações para o ponto de extremidade de destino. Você pode saber mais sobre monitoramento e métricas disponíveis na ferramenta de proxy.
Executar o carregamento de dados históricos
Para carregar os dados, crie um notebook Scala em sua conta do Azure Databricks. Substitua as configurações do Cassandra de origem e de destino pelas credenciais correspondentes e substitua os keyspaces e as tabelas de origem e de destino. Adicione mais variáveis para cada tabela conforme necessário para o exemplo a seguir e faça a execução. Depois que o aplicativo começar a enviar solicitações para o proxy de gravação dupla, estará tudo pronto para a migração de dados históricos.
Importante
Antes de migrar os dados, aumente a taxa de transferência do contêiner para o valor necessário para que seu aplicativo migre rapidamente. Dimensionar a taxa de transferência antes de iniciar a migração ajudará você a migrar seus dados em menos tempo. Para ajudar a proteger contra limitação de taxa durante a carga de dados histórica, talvez você queira habilitar as recuperações do lado do servidor (SSR) na API para Cassandra. Consulte nosso artigo aqui para obter mais informações e instruções sobre como habilitar o SSR.
import com.datastax.spark.connector._
import com.datastax.spark.connector.cql._
import org.apache.spark.SparkContext
// source cassandra configs
val sourceCassandra = Map(
"spark.cassandra.connection.host" -> "<Source Cassandra Host>",
"spark.cassandra.connection.port" -> "9042",
"spark.cassandra.auth.username" -> "<USERNAME>",
"spark.cassandra.auth.password" -> "<PASSWORD>",
"spark.cassandra.connection.ssl.enabled" -> "true",
"keyspace" -> "<KEYSPACE>",
"table" -> "<TABLE>"
)
//target cassandra configs
val targetCassandra = Map(
"spark.cassandra.connection.host" -> "<Source Cassandra Host>",
"spark.cassandra.connection.port" -> "10350",
"spark.cassandra.auth.username" -> "<USERNAME>",
"spark.cassandra.auth.password" -> "<PASSWORD>",
"spark.cassandra.connection.ssl.enabled" -> "true",
"keyspace" -> "<KEYSPACE>",
"table" -> "<TABLE>",
//throughput related settings below - tweak these depending on data volumes.
"spark.cassandra.output.batch.size.rows"-> "1",
"spark.cassandra.output.concurrent.writes" -> "1000",
"spark.cassandra.connection.remoteConnectionsPerExecutor" -> "1",
"spark.cassandra.concurrent.reads" -> "512",
"spark.cassandra.output.batch.grouping.buffer.size" -> "1000",
"spark.cassandra.connection.keep_alive_ms" -> "600000000"
)
//set timestamp to ensure it is before read job starts
val timestamp: Long = System.currentTimeMillis / 1000
//Read from source Cassandra
val DFfromSourceCassandra = sqlContext
.read
.format("org.apache.spark.sql.cassandra")
.options(sourceCassandra)
.load
//Write to target Cassandra
DFfromSourceCassandra
.write
.format("org.apache.spark.sql.cassandra")
.options(targetCassandra)
.option("writetime", timestamp)
.mode(SaveMode.Append)
.save
Observação
No exemplo Scala anterior, você observará que timestamp está sendo definido como a hora atual antes de ler todos os dados na tabela de origem. Em seguida, writetime será definido para esse carimbo de data/hora retroativo. Isso garante que os registros gravados do carregamento de dados históricos no ponto de extremidade de destino não consigam substituir as atualizações que chegam com um carimbo de data/hora posterior do proxy de gravação dupla enquanto os dados históricos estão sendo lidos.
Importante
Se você precisar preservar os carimbos de data/hora exatos por algum motivo, deverá adotar uma abordagem de migração de dados históricos que preserve os carimbos de data/hora, como este exemplo. O jar de dependência no exemplo também contém o conector do Spark, portanto, você não precisa instalar o assembly do conector Spark mencionado nos pré-requisitos anteriores – instalar ambos no cluster do Spark causará conflitos.
Validar a origem e o destino
Após o carregamento de dados históricos ser concluído, seus bancos de dados deverão estar em sincronia e prontos para a substituição. No entanto, é recomendável que você valide a origem e o destino para garantir que coincidam antes de fazer a substituição.
Observação
Se você usou o exemplo de migrador do Cassandra mencionado acima para preservar writetime, isso inclui a capacidade de validar a migraçãocomparando as linhas na origem e no destino com base em determinadas tolerâncias.