教學課程：使用 Spark 連線至 Azure Cosmos DB for NoSQL

適用於：NoSQL

在本教學課程中，您會使用 Azure Cosmos DB Spark 連接器，從 Azure Cosmos DB for NoSQL 帳戶讀取或寫入資料。 本教學課程使用 Azure Databricks 和 Jupyter 筆記本，藉此說明如何從 Spark 整合 API for NoSQL。 儘管您可以使用 Spark 支援的任何語言或介面，本教學課程著重於 Python 和 Scala。

在本教學課程中，您會了解如何：

• 使用 Spark 和 Jupyter 筆記本連線至 API for NoSQL 帳戶
• 建立資料庫和容器資源
• 將資料內嵌至容器
• 查詢容器中的資料
• 對容器中的項目執行一般作業

必要條件

• 現有的 Azure Cosmos DB for NoSQL 帳戶。
  • 如果您有現有的 Azure 訂用帳戶，請建立新的帳戶。
  • 沒有 Azure 訂用帳戶？ 您可以免費試用 Azure Cosmos DB，不需要信用卡。
• 現有 Azure Databricks 工作區。

使用 Spark 和 Jupyter 連線

使用現有的 Azure Databricks 工作區建立計算叢集，準備使用 Apache Spark 3.4.x 連線至 Azure Cosmos DB for NoSQL 帳戶。

1. 開啟您的 Azure Databricks 工作區。

2. 在工作區介面中，建立新的叢集。 設定叢集時，請至少符合下列設定：

   	值
   執行階段版本	13.3 LTS (Scala 2.12, Spark 3.4.1)

3. 使用工作區介面，從 [Maven Central] 搜尋群組識別碼為 com.azure.cosmos.spark 的 Maven 套件。 將成品識別碼前面為 azure-cosmos-spark_3-4 的 Spark 3.4 專用套件安裝至叢集。

4. 最後，建立新的筆記本。

   提示

   根據預設，筆記本會附加至最近建立的叢集。

5. 在筆記本中，設定 NoSQL 帳戶端點、資料庫名稱和容器名稱的 OLTP 組態設定。

   # Set configuration settings
   config = {
     "spark.cosmos.accountEndpoint": "<nosql-account-endpoint>",
     "spark.cosmos.accountKey": "<nosql-account-key>",
     "spark.cosmos.database": "cosmicworks",
     "spark.cosmos.container": "products"
   }
   

   # Set configuration settings
   val config = Map(
     "spark.cosmos.accountEndpoint" -> "<nosql-account-endpoint>",
     "spark.cosmos.accountKey" -> "<nosql-account-key>",
     "spark.cosmos.database" -> "cosmicworks",
     "spark.cosmos.container" -> "products"
   )
   

建立資料庫和容器

使用目錄 API 管理帳戶資源，例如資料庫和容器。 然後，您可以使用 OLTP 管理容器資源內的資料。

1. 使用 Spark 設定目錄 API 來管理 API for NoSQL 資源。

   # Configure Catalog Api    
   spark.conf.set("spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
   spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", config["spark.cosmos.accountEndpoint"])
   spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", config["spark.cosmos.accountKey"]) 
   

   // Configure Catalog Api  
   spark.conf.set(s"spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
   spark.conf.set(s"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", config("spark.cosmos.accountEndpoint"))
   spark.conf.set(s"spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", config("spark.cosmos.accountKey"))
   

2. 使用 CREATE DATABASE IF NOT EXISTS 建立名為 cosmicworks 的新資料庫。

   # Create a database using the Catalog API    
   spark.sql(f"CREATE DATABASE IF NOT EXISTS cosmosCatalog.cosmicworks;")
   

   // Create a database using the Catalog API  
   spark.sql(s"CREATE DATABASE IF NOT EXISTS cosmosCatalog.cosmicworks;")
   

3. 使用 CREATE TABLE IF NOT EXISTS 建立名為 products 的新容器。 確保將分割區索引鍵路徑設定為 /category，並啟用自動調整輸送量，最大輸送量為每秒 1000 要求單位 (RU/秒)。

   # Create a products container using the Catalog API
   spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.products USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/category', autoScaleMaxThroughput = '1000')"))
   

   // Create a products container using the Catalog API
   spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.products USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/category', autoScaleMaxThroughput = '1000')"))
   

4. 使用階層式分割區索引鍵設定，建立名為 employees 的另一個容器，其中 /organization、/department 和 /team 作為按該特定順序排列的分割區索引鍵路徑集。 此外，請將輸送量設定為手動數量 400 RU/秒

   # Create an employees container using the Catalog API
   spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.employees USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/organization,/department,/team', manualThroughput = '400')"))
   

   // Create an employees container using the Catalog API
   spark.sql(("CREATE TABLE IF NOT EXISTS cosmosCatalog.cosmicworks.employees USING cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/organization,/department,/team', manualThroughput = '400')"))
   

5. 執行筆記本資料格，以驗證資料庫和容器是否都在 API for NoSQL 帳戶內建立。

內嵌資料

建立範例資料集，然後使用 OLTP 將資料內嵌至 API for NoSQL 容器。

1. 建立範例資料集。

   # Create sample data    
   products = (
     ("68719518391", "gear-surf-surfboards", "Yamba Surfboard", 12, 850.00, False),
     ("68719518371", "gear-surf-surfboards", "Kiama Classic Surfboard", 25, 790.00, True)
   )
   

   // Create sample data
   val products = Seq(
     ("68719518391", "gear-surf-surfboards", "Yamba Surfboard", 12, 850.00, false),
     ("68719518371", "gear-surf-surfboards", "Kiama Classic Surfboard", 25, 790.00, true)
   )
   

2. 使用 spark.createDataFrame 和先前儲存的 OLTP 設定，將範例資料新增至目標容器。

   # Ingest sample data    
   spark.createDataFrame(products) \
     .toDF("id", "category", "name", "quantity", "price", "clearance") \
     .write \
     .format("cosmos.oltp") \
     .options(**config) \
     .mode("APPEND") \
     .save()
   

   // Ingest sample data
   spark.createDataFrame(products)
     .toDF("id", "category", "name", "quantity", "price", "clearance")
     .write
     .format("cosmos.oltp")
     .options(config)
     .mode("APPEND")
     .save()
   

查詢資料

將 OLTP 資料載入資料框架，以對資料執行一般查詢。 您可以使用各種語法篩選或查詢資料。

1. 使用 spark.read 將 OLTP 資料載入資料框架物件。 使用本教學課程稍早使用的相同設定。 此外，將 spark.cosmos.read.inferSchema.enabled 設為 true，以允許 Spark 連接器透過對現有項目取樣來推斷結構描述。

   # Load data    
   df = spark.read.format("cosmos.oltp") \
     .options(**config) \
     .option("spark.cosmos.read.inferSchema.enabled", "true") \
     .load()
   

   // Load data
   val df = spark.read.format("cosmos.oltp")
     .options(config)
     .option("spark.cosmos.read.inferSchema.enabled", "true")
     .load()
   

2. 使用 printSchema 轉譯資料框架中所載入資料的結構描述。

   # Render schema    
   df.printSchema()
   

   // Render schema    
   df.printSchema()
   

3. 轉譯資料列，其中 quantity 資料行小於 20。 使用 where 和 show 函式執行此查詢。

   # Render filtered data    
   df.where("quantity < 20") \
     .show()
   

   // Render filtered data
   df.where("quantity < 20")
     .show()
   

4. 轉譯 clearance 資料行為 true 的第一個資料列。 使用 filter 函式執行此查詢。

   # Render 1 row of flitered data    
   df.filter(df.clearance == True) \
     .show(1)
   

   // Render 1 row of flitered data
   df.filter($"clearance" === true)
     .show(1)
   

5. 轉譯五個沒有篩選或截斷的資料列。 使用 show 函式自訂轉譯的外觀和資料列數目。

   # Render five rows of unfiltered and untruncated data    
   df.show(5, False)
   

   // Render five rows of unfiltered and untruncated data    
   df.show(5, false)
   

6. 使用此原始 NoSQL 查詢字串查詢資料：SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800

   # Render results of raw query    
   rawQuery = "SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800"
   rawDf = spark.sql(rawQuery)
   rawDf.show()
   

   // Render results of raw query    
   val rawQuery = s"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE price > 800"
   val rawDf = spark.sql(rawQuery)
   rawDf.show()
   

執行一般作業

在 Spark 中使用 API for NoSQL 資料時，您可以執行部分更新，或以原始 JSON 形式使用資料。

1. 若要執行項目的部分更新，請執行下列步驟：

   1. 複製現有的 config 組態變數，並修改新複本中的屬性。 具體而言，將寫入策略設定為 ItemPatch，停用大量支援，設定資料行和對應作業，最後將預設作業類型設定為 Set。

      # Copy and modify configuration
      configPatch = dict(config)
      configPatch["spark.cosmos.write.strategy"] = "ItemPatch"
      configPatch["spark.cosmos.write.bulk.enabled"] = "false"
      configPatch["spark.cosmos.write.patch.defaultOperationType"] = "Set"
      configPatch["spark.cosmos.write.patch.columnConfigs"] = "[col(name).op(set)]"
      

      // Copy and modify configuration
      val configPatch = scala.collection.mutable.Map.empty ++ config
      configPatch ++= Map(
        "spark.cosmos.write.strategy" -> "ItemPatch",
        "spark.cosmos.write.bulk.enabled" -> "false",
        "spark.cosmos.write.patch.defaultOperationType" -> "Set",
        "spark.cosmos.write.patch.columnConfigs" -> "[col(name).op(set)]"
      )
      

   2. 針對您打算作為此修補作業目標之一的項目分割區索引鍵和唯一識別碼建立變數。

      # Specify target item id and partition key
      targetItemId = "68719518391"
      targetItemPartitionKey = "gear-surf-surfboards"
      

      // Specify target item id and partition key
      val targetItemId = "68719518391"
      val targetItemPartitionKey = "gear-surf-surfboards"
      

   3. 建立一組修補程式物件來指定目標項目，並指定應修改的欄位。

      # Create set of patch diffs
      patchProducts = [{ "id": f"{targetItemId}", "category": f"{targetItemPartitionKey}", "name": "Yamba New Surfboard" }]
      

      // Create set of patch diffs
      val patchProducts = Seq(
        (targetItemId, targetItemPartitionKey, "Yamba New Surfboard")
      )
      

   4. 使用該修補程式物件建立資料框架，並使用 write 執行修補作業。

      # Create data frame
      spark.createDataFrame(patchProducts) \
        .write \
        .format("cosmos.oltp") \
        .options(**configPatch) \
        .mode("APPEND") \
        .save()
      

      // Create data frame
      patchProducts
        .toDF("id", "category", "name")
        .write
        .format("cosmos.oltp")
        .options(configPatch)
        .mode("APPEND")
        .save()
      

   5. 執行查詢以檢閱修補作業的結果。 該項目現在應命名為 Yamba New Surfboard，沒有任何其他變更。

      # Create and run query
      patchQuery = f"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE id = '{targetItemId}' AND category = '{targetItemPartitionKey}'"
      patchDf = spark.sql(patchQuery)
      patchDf.show(1)
      

      // Create and run query
      val patchQuery = s"SELECT * FROM cosmosCatalog.cosmicworks.products WHERE id = '$targetItemId' AND category = '$targetItemPartitionKey'"
      val patchDf = spark.sql(patchQuery)
      patchDf.show(1)
      

2. 若要使用原始 JSON 資料，請執行下列步驟：

   1. 複製現有的 config 組態變數，並修改新複本中的屬性。 具體而言，將目標容器變更為 employees，並將 contacts 資料行/欄位設定為使用原始 JSON 資料。

      # Copy and modify configuration
      configRawJson = dict(config)
      configRawJson["spark.cosmos.container"] = "employees"
      configRawJson["spark.cosmos.write.patch.columnConfigs"] = "[col(contacts).path(/contacts).op(set).rawJson]"
      

      // Copy and modify configuration
      val configRawJson = scala.collection.mutable.Map.empty ++ config
      configRawJson ++= Map(
        "spark.cosmos.container" -> "employees",
        "spark.cosmos.write.patch.columnConfigs" -> "[col(contacts).path(/contacts).op(set).rawJson]"
      )
      

   2. 建立一組員工以內嵌至容器。

      # Create employee data
      employees = (
        ("63476388581", "CosmicWorks", "Marketing", "Outside Sales", "Alain Henry",  '[ { "type": "phone", "value": "425-555-0117" }, { "email": "alain@adventure-works.com" } ]'), 
      )
      

      // Create employee data
      val employees = Seq(
        ("63476388581", "CosmicWorks", "Marketing", "Outside Sales", "Alain Henry",  """[ { "type": "phone", "value": "425-555-0117" }, { "email": "alain@adventure-works.com" } ]""")
      )
      

   3. 建立資料框架，並使用 write 內嵌員工資料。

      # Ingest data
      spark.createDataFrame(employees) \
        .toDF("id", "organization", "department", "team", "name", "contacts") \
        .write \
        .format("cosmos.oltp") \
        .options(**configRawJson) \
        .mode("APPEND") \
        .save()
      

      // Ingest data
      spark.createDataFrame(employees)
        .toDF("id", "organization", "department", "team", "name", "contacts")
        .write
        .format("cosmos.oltp")
        .options(configRawJson)
        .mode("APPEND")
        .save()
      

   4. 使用 show 從資料框架轉譯資料。 觀察輸出中的 contacts 資料行是原始 JSON。

      # Read and render data
      rawJsonDf = spark.read.format("cosmos.oltp") \
        .options(**configRawJson) \
        .load()
      rawJsonDf.show()
      

      // Read and render data
      val rawJsonDf = spark.read.format("cosmos.oltp")
        .options(configRawJson)
        .load()
      rawJsonDf.show()
      

相關內容

• Apache Spark
• Azure Cosmos DB 目錄 API
• 設定參數參考
• 範例「紐約市計程車資料」筆記本
• 從 Spark 2.4 移轉至 Spark 3.*
• 版本相容性
  • Spark 3.1 的版本相容性
  • Spark 3.2 的版本相容性
  • Spark 3.3 的版本相容性
  • Spark 3.4 的版本相容性
• 版本資訊
  • Spark 3.1 的版本資訊
  • Spark 3.2 的版本資訊
  • Spark 3.3 的版本資訊
  • Spark 3.4 的版本資訊
• 下載連結
  • 下載適用於 Spark 3.1 的 Azure Cosmos DB Spark 連接器
  • 下載適用於 Spark 3.2 的 Azure Cosmos DB Spark 連接器
  • 下載適用於 Spark 3.3 的 Azure Cosmos DB Spark 連接器
  • 下載適用於 Spark 3.4 的 Azure Cosmos DB Spark 連接器

後續步驟

Maven 中央存放庫上的 Azure Cosmos DB Spark 連接器