使用 Azure Databricks 將資料從 Cassandra 移轉至 Azure Cosmos DB for Apache Cassandra 帳戶

適用於: ✅ Apache Cassandra

適用於：卡珊德拉

Important

您是否正在尋找一種適用於高擴展性場景的資料庫解決方案，且具有 99.999% 可用性的服務等級協定（SLA）、即時自動擴展，以及跨多個區域的自動容錯切換？考慮使用Azure Cosmos DB作為NoSQL的選擇。

您是否想要移轉現有的 Apache Cassandra 應用程式？請考慮 Azure 受控執行個體 for Apache Cassandra。

因為數個原因，Azure Cosmos DB 中的 API for Cassandra 已成為在 Apache Cassandra 上執行企業工作負載的絕佳選擇：

沒有管理和監視額外負荷：可消除跨作業系統、JVM 和 YAML 檔案和其互動的管理和監視設定額外負荷。
大幅節省成本：您可以使用 Azure Cosmos DB 來節省成本，包括 VM、頻寬和任何適用授權的成本。您不需要管理資料中心、伺服器、SSD 儲存體、網路和電力成本。
可使用現有的程式碼和工具：Azure Cosmos DB 提供與現有 Cassandra SDK 和工具相容的有線通訊協定層級。此相容性確保您可以透過 Azure Cosmos DB for Apache Cassandra 使用現有程式碼基底來執行瑣碎的變更。

有許多方法可以將資料庫工作負載從某個平台移轉至另一個平台。 Azure Databricks 是適用於 Apache Spark 的平台即服務 (PaaS) 供應項目，可提供方法來大規模執行離線移轉。本文描述使用 Azure Databricks 將資料從原生 Apache Cassandra keyspace 和資料表移轉至 Azure Cosmos DB for Apache Cassandra 所需的步驟。

Prerequisites

佈建 Azure Cosmos DB for Apache Cassandra 帳戶。
檢視連線至 Azure Cosmos DB for Apache Cassandra 的基本概念。
請檢閱 Azure Cosmos DB for Apache Cassandra 中支援的功能，以確保相容性。
確定您已在目標 Azure Cosmos DB for Apache Cassandra 帳戶中建立空的 keyspace 和資料表。
使用 cqlsh 進行驗證。

佈建 Azure Databricks 工作區

你可以依照指示配置 Azure Databricks 工作空間。你可以使用預設的運算資源，或建立新的運算資源來執行你的筆記本。務必選擇至少支援 Spark 3.0 的 Databricks 執行環境。

新增依賴性

你需要將 Apache Spark Cassandra Connector 函式庫加入你的運算系統，才能同時連接原生和 Azure Cosmos DB Cassandra 端點。在你的計算環境中，選擇 「Libraries>Install New>Maven」，然後在 Maven 座標中添加 com.datastax.spark:spark-cassandra-connector-assembly_2.12:3.0.0。

選擇安裝，安裝完成後重新啟動運算。

Note

安裝 Cassandra Connector 函式庫後，務必重新啟動 Databricks 的運算。

Warning

本文所示的範例已與 Spark 3.0.1 版本和對應的 Cassandra Spark 連接器 com.datastax.spark:spark-cassandra-connector-assembly_2.12:3.0.0 一同測試。較新版本的 Spark 和/或 Cassandra 連接器可能無法如預期般運作。

建立 Scala 筆記本以進行移轉

在 Databricks 中建立 Scala 筆記本。將您的來源和目標 Cassandra 設定取代為對應的憑證，以及來源和目標 keyspace 區域和資料表。然後執行下列程式碼：

import com.datastax.spark.connector._
import com.datastax.spark.connector.cql._
import org.apache.spark.SparkContext

// source cassandra configs
val nativeCassandra = 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 cosmosCassandra = Map( 
    "spark.cassandra.connection.host" -> "<USERNAME>.cassandra.cosmos.azure.com",
    "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 3.x
    "spark.cassandra.connection.connections_per_executor_max"-> "1", // Spark 2.x
    "spark.cassandra.concurrent.reads" -> "512",
    "spark.cassandra.output.batch.grouping.buffer.size" -> "1000",
    "spark.cassandra.connection.keep_alive_ms" -> "600000000"
)

//Read from native Cassandra
val DFfromNativeCassandra = sqlContext
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(nativeCassandra)
  .load
  
//Write to CosmosCassandra
DFfromNativeCassandra
  .write
  .format("org.apache.spark.sql.cassandra")
  .options(cosmosCassandra)
  .mode(SaveMode.Append) // only required for Spark 3.x
  .save

Note

spark.cassandra.output.batch.size.rows 和 spark.cassandra.output.concurrent.writes 值以及 Spark 叢集中的背景工作數目是為了避免速率限制而需要微調的重要設定。 Azure Cosmos DB 的要求超過佈建輸送量或要求單位 (RU) 時，就會發生速率限制。根據 Spark 叢集中的執行緒數量，以及可能的寫入目標資料表的每筆記錄大小（因此也影響 RU 成本），您可能需要調整這些設定。

Troubleshoot

頻率限制 (429 錯誤)

即使您將設定減少為最小值，可能還是會看到 429 錯誤碼或「要求速率太大」錯誤文字。下列案例可能會導致速率限制：

配置給資料表的輸送量小於 6,000 個要求單位。即使在最低設定下，Spark 還是可以以每秒大約 6,000 或更多要求單位的速率進行寫入。如果您已在具有共用吞吐量的 keyspace 中設置了一張資料表，則這個資料表在執行時可能只有不到 6,000 個請求單位（RU）可用。

執行移轉時，請確定您要移轉至其中的資料表至少有 6,000 個 RU 可用。如有必要，請將專用請求單位配置給該資料表。
大量資料量導致過度的資料傾斜。如果您有大量資料要移轉至指定的資料表，但在資料中有明顯的扭曲 (即針對相同的分割區索引鍵值寫入大量記錄)，則即使您在資料表中佈建數個要求單位，仍然可能會遇到速率限制。請求單位會在實體分割區中平均分配，而嚴重的資料不均可能會造成單一分割區請求的瓶頸。

在此案例中，請將 Spark 的輸送量設置調整至最低，並強制遷移緩慢執行。如果您要移轉參考或控制資料表，但存取頻率較低且扭曲可能很高，則此案例可能較常見。不過，如果任何其他類型的資料表中有明顯的扭曲，則建議您在穩定狀態期間檢閱您的資料模型，以避免工作負載的常用分割區問題。

下一步

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-04-27