Öğretici: Spark OLTP bağlayıcısı ile Delta Lake'ten NoSQL için Azure Cosmos DB'ye ters ayıklama, dönüştürme ve yükleme (ETL)

Bu öğreticide, Zenginleştirilmiş verileri Azure Databricks'teki Delta tablolarından NoSQL için Azure Cosmos DB'ye taşımak için ters ETL işlem hattı ayarlarsınız. Ardından, verileri eşitlemek için NoSQL için Azure Cosmos DB için Çevrimiçi İşlem İşleme (OLTP) Spark bağlayıcısını kullanırsınız.

Ters ETL işlem hattı kurulumu önkoşulları

• Mevcut bir Azure Cosmos DB hesabı.
  • Azure aboneliğiniz varsa yeni bir hesap oluşturun.
• Mevcut bir Azure Databricks çalışma alanı.
  • Azure aboneliğiniz varsa yeni bir çalışma alanı oluşturun.
• Azure CLI'nın en son sürümü.
  • İsterseniz Azure Cloud Shell'i de kullanabilirsiniz.

Microsoft Entra ile rol tabanlı erişim denetimini yapılandırma

Azure yönetilen kimlikleri, kimlik bilgilerini el ile yönetmeden NoSQL için Azure Cosmos DB'de güvenli ve parolasız kimlik doğrulaması sağlar. Bu önkoşul adımında, Azure Databricks'in meta verilere okuma erişimi ve NoSQL için Azure Cosmos DB hesabınız için verilere yazma erişimi ile otomatik olarak oluşturduğu kullanıcı tarafından atanan yönetilen kimliği ayarlayın. Bu adım, yönetilen kimlik için hem denetim hem de veri düzlemi rol tabanlı erişim denetimi rollerini yapılandırmaktadır.

1. Azure portalında (https://portal.azure.com) oturum açın.

2. Mevcut Azure Databricks kaynağına gidin.

3. Temel Bileşenler bölmesinde çalışma alanıyla ilişkilendirilmiş yönetilen kaynak grubunu bulun ve bu gruba gidin.

4. Yönetilen kaynak grubunda, çalışma alanıyla otomatik olarak oluşturulan kullanıcı tarafından atanan yönetilen kimliği seçin.

5. temel bileşenler bölmesine İstemci Kimliği ve Nesne (sorumlu) kimliği alanlarının değerini kaydedin. Bu değeri daha sonra denetim ve veri düzlemi rolleri atamak için kullanırsınız.

   Tip

   Alternatif olarak, Azure CLI kullanarak yönetilen kimliğin asıl kimliğini alabilirsiniz. Yönetilen kimliğin adının dbmanagedidentity olduğunu varsayarsak, asıl kimlik kimliği almak için az resource show komutunu kullanın.

   az resource show \
       --resource-group "<name-of-managed-resource-group>" \
       --name "dbmanagedidentity" \
       --resource-type "Microsoft.ManagedIdentity/userAssignedIdentities" \
       --query "{clientId: properties.clientId, principalId: properties.principalId}"
   

6. NoSQL için hedef Azure Cosmos DB hesabına gidin.

7. Hesabın sayfasında Erişim denetimi (IAM) öğesini seçin.

8. Erişim denetimi bölmesinde Ekle'yi ve ardından Rol ataması ekle seçeneklerini seçerek kullanıcı tarafından atanan yönetilen kimliğe bir denetim düzlemi rolü atama işlemini başlatın.

9. Atama rolleri listesinde Cosmos DB Hesap Okuyucusu rolünü seçin.

10. Bir kullanıcıya, gruba veya hizmet sorumlusuna erişim atama bölümünde üyeleri seç seçeneğiyle etkileşime geçin.

11. Üyeler iletişim kutusunda, Azure Databricks ile ilişkilendirilmiş kullanıcı tarafından atanan yönetilen kimliğe filtre uygulamak için asıl kimliği girin. Bu kimliği seçin .

12. Son olarak, denetim düzlemi rol atamasını oluşturmak için Gözden Geçir ve Ata seçin.

13. Kullanıcı tarafından atanan, Azure Databricks ile ilişkili yönetilen kimliğe veri düzlemi rolü az cosmosdb sql role assignment create ve kapsamı Cosmos DB Built-in Data Contributor atamak için / komutunu kullanın.

    az cosmosdb sql role assignment create \
        --resource-group "<name-of-resource-group>" \
        --account-name "<name-of-cosmos-nosql-account>" \
        --principal-id "<managed-identity-principal-id>" \
        --role-definition-name "Cosmos DB Built-in Data Contributor" \ --scope "/"
    

14. Aboneliğinizi ve kiracı tanımlayıcılarınızı almak için kullanın az account show . Bu değerler, Microsoft Entra kimlik doğrulaması kullanılarak Spark bağlayıcısı ile sonraki bir adımda gereklidir.

    az account show --query '{subscriptionId: id, tenantId: tenantId}'
    

Databricks not defteri oluşturma

1. Mevcut Azure Databricks kaynağına gidin ve çalışma alanı kullanıcı arabirimini açın.

2. Henüz bir kümeniz yoksa yeni bir küme oluşturun.

   Important

   Kümenin Spark 3.5.0 ve Scala 2.12 için uzun vadeli desteğe sahip olan Çalışma Zamanı sürüm 15.4 ya da üzeri olduğundan emin olun. Bu kılavuzdaki diğer adımlarda araçların bu sürümleri varsayılır.

3. Maven paketini yüklemek için Kitaplıklar>Yeni Yükle> ve Maven'e gidin.

4. Grup Kimliği filtresini com.azure.cosmos.spark kullanarak ve yapıt kimliğine sahip paketi seçerek NoSQL için Azure Cosmos DB için Spark bağlayıcısını azure-cosmos-spark_3-5_2-12arayın.

5. Çalışma Alanı>[Klasör]>Yeni>Not Defteri seçeneğine giderek yeni bir not defteri oluşturun.

6. Not defterini kümenize ekleyin.

Azure Databricks'te Spark bağlayıcısı yapılandırma

Microsoft Entra kimlik doğrulamasını kullanarak Spark bağlayıcısını hesabınızın kapsayıcısına bağlanacak şekilde yapılandırın. Ayrıca, bağlayıcıyı Spark işlemleri için yalnızca sınırlı bir aktarım hızı eşiği kullanacak şekilde yapılandırın. Spark bağlayıcısını yapılandırmak için, hesabınıza bağlanmak için kimlik bilgileriyle bir yapılandırma sözlüğü tanımlayın. Bu kimlik bilgileri şunlardır:

cosmos_config = {
    # General settings
    "spark.cosmos.accountEndpoint": "<endpoint>",
    "spark.cosmos.database": "products",
    "spark.cosmos.container": "recommendations",
    # Entra authentication settings
    "spark.cosmos.auth.type": "ManagedIdentity",
    "spark.cosmos.account.subscriptionId": "<subscriptionId>",
    "spark.cosmos.account.tenantId": "<tenantId>",
    "spark.cosmos.account.resourceGroupName": "<resourceGroupName>",
    # Throughput control settings
    "spark.cosmos.throughputControl.enabled": "true",
    "spark.cosmos.throughputControl.name": "TargetContainerThroughputControl",
    "spark.cosmos.throughputControl.targetThroughputThreshold": "0.30",
    "spark.cosmos.throughputControl.globalControl.useDedicatedContainer": "false",
}

val cosmosconfig = Map(
  // General settings
  "spark.cosmos.accountEndpoint" -> "<endpoint>",
  "spark.cosmos.database" -> "products",
  "spark.cosmos.container" -> "recommendations",
  // Entra authentication settings
  "spark.cosmos.auth.type" -> "ManagedIdentity",
  "spark.cosmos.account.subscriptionId" -> "<subscriptionId>",
  "spark.cosmos.account.tenantId" -> "<tenantId>",
  "spark.cosmos.account.resourceGroupName" -> "<resourceGroupName>",
  // Throughput control settings
  "spark.cosmos.throughputControl.enabled" -> "true",
  "spark.cosmos.throughputControl.name" -> "TargetContainerThroughputControl",
  "spark.cosmos.throughputControl.targetThroughputThreshold" -> "0.30",
  "spark.cosmos.throughputControl.globalControl.useDedicatedContainer" -> "false"
)

Bu adımda belirtildiği gibi aktarım hızı yapılandırması, Spark işlemleri için hedef kapsayıcıya ayrılan istek birimlerinin (RU) yalnızca 30% kullanılabilir olmasını sağlar.

Delta tablosuna örnek ürün önerileri verilerini alma

Kullanıcılar için ürün önerileri bilgilerini içeren örnek bir DataFrame oluşturun ve adlı recommendations_deltabir Delta tablosuna yazın. Bu adım, NoSQL uyumlu Azure Cosmos DB ile eşitlemeyi planladığınız veri gölünüzdeki seçilmiş ve dönüştürülmüş verilerin simülasyonunu yapar. Delta biçimine yazmak, daha sonra artımlı eşitleme için değişiklik veri yakalamayı (CDC) etkinleştirebilmenizi sağlar.

from pyspark.sql import SparkSession

# Create sample data and convert it to a DataFrame
df = spark.createDataFrame([
    ("yara-lima", "Full-Finger Gloves", "clothing-gloves", 80),
    ("elza-pereira", "Long-Sleeve Logo Jersey", "clothing-jerseys", 90)
], ["id", "productname", "category", "recommendationscore"])

# Write the DataFrame to a Delta table
df.write.mode("append").format("delta").saveAsTable("recommendations_delta")

// Create sample data as a sequence and convert it to a DataFrame
val df = Seq(
  ("yara-lima", "Full-Finger Gloves", "clothing-gloves", 12.95),
  ("elza-pereira", "Long-Sleeve Logo Jersey", "clothing-jerseys", 19.99)
).toDF("id", "productname", "category", "recommendationscore") 

// Write the DataFrame to a table
df.write.mode("append").format("delta").saveAsTable("recommendations_delta")

NoSQL için Azure Cosmos DB'ye ilk verileri toplu yükleme

Sonra, Delta tablosunu bir Spark DataFrame'e okuyun ve recommendations_delta biçimini kullanarak, NoSQL için Azure Cosmos DB'ye ilk toplu yazma işlemini gerçekleştirin cosmos.oltp. Hedef veritabanı ve kapsayıcıdaki mevcut içeriğin üzerine yazmadan verileri eklemek için ekleme modunu kullanın. Bu adım, CDC başlamadan önce tüm geçmiş verilerin hesapta kullanılabilir olmasını sağlar.

# Read the Delta table into a DataFrame
df_delta = spark.read.format("delta").table("recommendations_delta")

# Write the DataFrame to the container using the Cosmos OLTP format
df_delta.write.format("cosmos.oltp").mode("append").options(**cosmos_config).save()

// Read the Delta table into a DataFrame
val df_delta = spark.read.format("delta").table("recommendations_delta")

// Write the DataFrame to the container using the Cosmos OLTP format
df_delta.write.format("cosmos.oltp").mode("append").options(cosmosconfig).save()

Değişiklik veri beslemesiyle akış senkronizasyonunu etkinleştirme

Tablonun özelliklerini değiştirerek Delta Lake'in Veri Akışını Değiştir (CDF) özelliğini recommendations_delta tabloda etkinleştirin. CDF, Delta Lake'in gelecekteki tüm satır düzeyi eklemeleri, güncelleştirmeleri ve silmeleri izlemesine olanak tanır. Bu özelliğin etkinleştirilmesi, anlık görüntüleri karşılaştırmaya gerek kalmadan değişiklikleri kullanıma sunmasından dolayı NoSQL için Azure Cosmos DB'de artımlı eşitlemeler gerçekleştirmek için gereklidir.

Geçmiş veri yükünden sonra Delta tablosundaki değişiklikler Delta Değişiklik Veri Akışı (CDF) kullanılarak yakalanabilir. Toplu işlem odaklı veya akış odaklı CDC uygulayabilirsiniz.

# Enable Change Data Feed (CDF)
spark.sql("""
  ALTER TABLE recommendations_delta SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# Read the Change Data Capture (CDC) data from the Delta table
cdc_batch_df = spark.read.format("delta").option("readChangeData", "true").option("startingVersion", "1").table("recommendations_delta")

# Write the captured changes to Azure Cosmos DB for NoSQL in append mode
cdc_batch_df.write.format("cosmos.oltp").mode("append").options(**cosmos_config).save()

# Enable Change Data Feed (CDF)
spark.sql("""
  ALTER TABLE recommendations_delta SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# Read Change Data Capture (CDC) stream from the Delta table 'recommendations_delta'
cdc_stream_df = spark.readStream.format("delta").option("readChangeData", "true").table("recommendations_delta")

# Write the CDC stream to Azure Cosmos DB for NOSQL using the OLTP connector with checkpointing
streaming_query = cdc_stream_df.writeStream.format("cosmos.oltp").outputMode("append").options(**cosmos_config).option("checkpointLocation", "/mnt/checkpoints/recommendations-stream").start()

# Wait for the stream to terminate or time out after 60 seconds
try:   streaming_query.awaitTermination(60)
except:     print("Stream stopped or timed out.")

# Stop the stream if it’s still running after the timeout
if streaming_query.isActive: streaming_query.stop()

// Enable Change Data Feed (CDF)
spark.sql("""
  ALTER TABLE recommendations_delta SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

// Read the Change Data Capture (CDC) data from the Delta table
val cdc_batch_df = spark.read.format("delta").option("readChangeData", "true").option("startingVersion", "1").table("recommendations_delta")

// Write the captured changes to Azure Cosmos DB for NoSQL in append mode
cdc_batch_df.write.format("cosmos.oltp").mode("append").options(cosmos_config).save()

// Enable Change Data Feed (CDF)
spark.sql("""
  ALTER TABLE recommendations_delta SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

// Read Change Data Capture (CDC) stream from the Delta table 'recommendations_delta'
val cdcStreamDF = spark.readStream
  .format("delta")
  .option("readChangeData", "true")
  .table("recommendations_delta")

// Write the CDC stream to Azure Cosmos DB for NoSQL using the OLTP connector with checkpointing
val streamingQuery = cdcStreamDF.writeStream
  .format("cosmos.oltp")
  .outputMode("append")
  .options(cosmosconfig)
  .option("checkpointLocation", "/mnt/checkpoints/recommendations-stream")
  .start()

// Wait for the stream to terminate or time out after 60 seconds
try {
  streamingQuery.awaitTermination(60000) // time out in milli seconds
} catch {
  case e: Exception => println("Stream stopped or timed out.")
}
// Stop the stream if it’s still running after the timeout
if (streamingQuery.isActive) {
  streamingQuery.stop()
}

NoSQL sorgularını kullanarak verileri doğrulama

NoSQL için Azure Cosmos DB'ye yazdıktan sonra aynı hesap yapılandırmasını kullanarak verileri Spark'a sorgulayarak doğrulayın. Sonra; Alınan verileri inceleyin, doğrulamaları çalıştırın veya analiz veya raporlama için Delta Lake'teki diğer veri kümelerine katılın. NoSQL için Azure Cosmos DB, gerçek zamanlı sorgu performansı için hızlı, dizinli okumaları destekler.

# Load DataFrame
df_cosmos = spark.read.format("cosmos.oltp").options(**cosmos_config).load()

# Run query
df_cosmos.select("id", "productname", "category", "recommendationscore").show()

// Load DataFrame
val dfCosmos = spark.read.format("cosmos.oltp").options(cosmosConfig).load()

// Run query
dfCosmos.select("id", "productname", "category", "recommendationscore").show()

İlgili içerik

• ETL’in Tersine Çevrilmesi (Reverse ETL)
• Azure Cosmos DB Spark Bağlayıcısı
• Apache Spark'ta Microsoft Entra kimlik doğrulaması