Migrálás az Apache Cassandra felügyelt Azure-példányára az Apache Spark használatával
Ahol lehetséges, javasoljuk, hogy az Apache Cassandra natív replikációval migrálja az adatokat a meglévő fürtből az Apache Cassandra felügyelt Azure-példányába egy hibrid fürt konfigurálásával. Ez a módszer az Apache Cassandra pletykaprotokollját fogja használni az adatok replikálására a forrásadatközpontból az új felügyelt példány adatközpontjába. Előfordulhat azonban, hogy a forrásadatbázis verziója nem kompatibilis, vagy a hibrid fürt beállítása egyébként nem valósítható meg.
Ez az oktatóanyag bemutatja, hogyan migrálhatja az adatokat az Apache Cassandra felügyelt Azure-példányára offline módon a Cassandra Spark Connector és az Azure Databricks for Apache Spark használatával.
Előfeltételek
Felügyelt Azure-példány kiépítése Apache Cassandra-fürthöz Azure Portal vagy az Azure CLI használatával, és győződjön meg arról, hogy cQLSH-val tud csatlakozni a fürthöz.
Azure Databricks-fiók kiépítése a felügyelt Cassandra virtuális hálózaton belül. Győződjön meg arról, hogy hálózati hozzáféréssel is rendelkezik a forrás Cassandra-fürthöz.
Győződjön meg arról, hogy már migrálta a keyspace/table sémát a forrás Cassandra-adatbázisból a cél cassandrai felügyeltpéldány-adatbázisba.
Azure Databricks-fürt kiépítése
Javasoljuk, hogy válassza a Databricks futtatókörnyezet 7.5-ös verzióját, amely támogatja a Spark 3.0-t.
Függőségek hozzáadása
Adja hozzá az Apache Spark Cassandra Connector kódtárat a fürthöz, hogy natív és Azure Cosmos DB Cassandra-végpontokhoz is csatlakozzon. A fürtben válassza a Kódtárak>Új>Maven telepítése lehetőséget, majd adja hozzá com.datastax.spark:spark-cassandra-connector-assembly_2.12:3.0.0 a Maven koordinátáit.
Válassza a Telepítés lehetőséget, majd a telepítés befejezése után indítsa újra a fürtöt.
Megjegyzés
A Cassandra Connector-kódtár telepítése után indítsa újra a Databricks-fürtöt.
Scala-jegyzetfüzet létrehozása migráláshoz
Scala-jegyzetfüzet létrehozása a Databricksben. Cserélje le a forrás- és cél Cassandra-konfigurációkat a megfelelő hitelesítő adatokra, valamint a forrás- és célkulcsterekre és -táblákra. Ezután futtassa a következő kódot:
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" -> "false",
"keyspace" -> "<KEYSPACE>",
"table" -> "<TABLE>"
)
//target cassandra configs
val targetCassandra = 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>",
//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" -> "10",
"spark.cassandra.concurrent.reads" -> "512",
"spark.cassandra.output.batch.grouping.buffer.size" -> "1000",
"spark.cassandra.connection.keep_alive_ms" -> "600000000"
)
//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)
.mode(SaveMode.Append) // only required for Spark 3.x
.save
Megjegyzés
Ha meg kell őriznie az egyes sorok eredeti writetime részét, tekintse meg a cassandra-migrátormintát .