Aracılığıyla paylaş

Azure IoT MQ Preview'dan Data Lake'e veri gönderme Depolama

  • Makale

Önemli

Azure Arc tarafından etkinleştirilen Azure IoT İşlemleri Önizlemesi şu anda ÖNİzLEME aşamasındadır. Bu önizleme yazılımını üretim ortamlarında kullanmamalısınız.

Beta veya önizleme aşamasında olan ya da başka bir şekilde henüz genel kullanıma sunulmamış olan Azure özelliklerinde geçerli olan yasal koşullar için bkz. Microsoft Azure Önizlemeleri için Ek Kullanım Koşulları.

Data Lake bağlayıcısını kullanarak Azure IoT MQ Preview aracısından Azure Data Lake Storage 2. Nesil (ADLSv2), Microsoft Fabric OneLake ve Azure Veri Gezgini gibi bir veri gölüne veri gönderebilirsiniz. Bağlayıcı MQTT konularına abone olur ve iletileri Data Lake Depolama hesabındaki Delta tablolarına alır.

Önkoşullar

Yönetilen kimlik kullanarak Microsoft Fabric OneLake'e veri gönderecek şekilde yapılandırma

Yönetilen kimlik kullanarak Microsoft Fabric OneLake'e bağlanmak için bir data lake bağlayıcısı yapılandırın.

  1. Microsoft Fabric çalışma alanı ve lakehouse dahil olmak üzere önkoşullardaki adımların karşılandığından emin olun. Varsayılan çalışma alanım kullanılamaz.

  2. IoT MQ Arc uzantısının yüklü olduğundan ve yönetilen kimlikle yapılandırıldığından emin olun.

  3. Azure portalında Arc bağlantılı Kubernetes kümesine gidin ve Ayarlar> Extensions'ı seçin. Uzantı listesinde IoT MQ uzantınızın adını arayın. Ad, ardından beş rastgele karakterle mq- başlar. Örneğin, mq-4jgjs.

  4. IoT MQ Arc uzantısı yönetilen kimliğiyle ilişkili uygulama kimliğini alın ve GUID değerini not edin. Uygulama kimliği nesneden veya asıl kimlikten farklıdır. Yönetilen kimliğin nesne kimliğini bulup yönetilen kimlikle ilişkili hizmet sorumlusunun uygulama kimliğini sorgulayarak Azure CLI'yi kullanabilirsiniz. Örneğin:

    OBJECT_ID=$(az k8s-extension show --name <IOT_MQ_EXTENSION_NAME> --cluster-name <ARC_CLUSTER_NAME> --resource-group <RESOURCE_GROUP_NAME> --cluster-type connectedClusters --query identity.principalId -o tsv)
az ad sp show --query appId --id $OBJECT_ID --output tsv

    GUID değerine sahip bir çıkış almanız gerekir:

    xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

    Bu GUID, bir sonraki adımda kullanmanız gereken uygulama kimliğidir.

  5. Microsoft Fabric çalışma alanında Erişimi yönet'i kullanın, ardından + Kişi veya grup ekle'yi seçin.

  6. IoT MQ Arc uzantısını "mq" adıyla arayın ve önceki adımda bulduğunuz uygulama kimliği GUID değerini seçtiğinizden emin olun.

  7. Rol olarak Katkıda Bulunan'ı ve ardından Ekle'yi seçin.

  8. Bağlayıcı için yapılandırma ve uç nokta ayarlarını tanımlayan bir DataLake Bağlan or kaynağı oluşturun. Örnek olarak sağlanan YAML'yi kullanabilirsiniz, ancak aşağıdaki alanları değiştirdiğinizden emin olun:

    • target.fabricOneLake.endpoint: Microsoft Fabric OneLake hesabının uç noktası. Uç nokta URL'sini Dosya>Özellikleri altında Microsoft Fabric lakehouse'dan alabilirsiniz. URL gibi https://onelake.dfs.fabric.microsoft.comgörünmelidir.
    • target.fabricOneLake.names: Çalışma alanı ve göl evi adları. Bu alanı veya guidskullanın. İkisini birden kullanmayın.
      • workspaceName: Çalışma alanının adı.
      • lakehouseName: Göl evi adı.
    apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnector
metadata:
  name: my-datalake-connector
  namespace: azure-iot-operations
spec:
  protocol: v5
  image:
    repository: mcr.microsoft.com/azureiotoperations/datalake
    tag: 0.4.0-preview
    pullPolicy: IfNotPresent
  instances: 2
  logLevel: info
  databaseFormat: delta
  target:
    fabricOneLake:
      # Example: https://onelake.dfs.fabric.microsoft.com
      endpoint: <example-endpoint-url>
      names:
        workspaceName: <example-workspace-name>
        lakehouseName: <example-lakehouse-name>
      ## OR
      # guids:
      #   workspaceGuid: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
      #   lakehouseGuid: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
      fabricPath: tables
      authentication:
        systemAssignedManagedIdentity:
          audience: https://storage.azure.com/
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}

  9. Data Lake Depolama MQTT konusu ile Delta tablosu arasındaki eşlemeyi tanımlayan bir DataLake Bağlan orTopicMap kaynağı oluşturun. Örnek olarak sağlanan YAML'yi kullanabilirsiniz, ancak aşağıdaki alanları değiştirdiğinizden emin olun:

    • dataLakeConnectorRef: Daha önce oluşturduğunuz DataLake Bağlan or kaynağının adı.
    • clientId: MQTT istemciniz için benzersiz bir tanımlayıcı.
    • mqttSourceTopic: Verilerin gelmesini istediğiniz MQTT konusunun adı.
    • table.tableName: Göl evinde eklemek istediğiniz tablonun adı. Tablo yoksa otomatik olarak oluşturulur.
    • table.schema: MQTT konusuna gönderdiğiniz JSON iletilerinin biçimi ve alanlarıyla eşleşmesi gereken Delta tablosunun şeması.

  10. Kullanarak kubectl apply -f datalake-connector.yamlKubernetes kümenize DataLake Bağlan or ve DataLake Bağlan orTopicMap kaynaklarını uygulayın.

  11. MQTT yayımcınızı kullanarak MQTT konusuna JSON iletileri göndermeye başlayın. Data Lake bağlayıcı örneği konuya abone olur ve iletileri Delta tablosuna alır.

  12. Tarayıcı kullanarak verilerin lakehouse'a aktarıldığını doğrulayın. Microsoft Fabric çalışma alanında göl evinizi ve ardından Tablolar'ı seçin. Verileri tabloda görmeniz gerekir.

Tanımlanamayan tablo

Verileriniz Tanımlanamayan tabloda gösteriliyorsa :

Neden, tablo adında desteklenmeyen karakterler olabilir. Tablo adı geçerli bir Azure Depolama kapsayıcı adı olmalıdır; bu, uzunluğu 256 karaktere kadar olan herhangi bir İngilizce harfi( büyük veya küçük harf) ve alt çubuk _içerebileceği anlamına gelir. Tire - veya boşluk karakterine izin verilmez.

SAS belirtecini kullanarak Azure Data Lake Storage 2. Nesil veri gönderecek şekilde yapılandırma

Data Lake bağlayıcısını, paylaşılan erişim imzası (SAS) belirtecini kullanarak bir Azure Data Lake Storage 2. Nesil (ADLS 2. Nesil) hesabına bağlanacak şekilde yapılandırın.

  1. Azure Data Lake Storage 2. Nesil (ADLS 2. Nesil) hesabı için SAS belirteci alın. Örneğin, depolama hesabınıza göz atmak için Azure portalını kullanın. Güvenlik + ağ altındaki menüde Paylaşılan erişim imzası'nı seçin. Gerekli izinleri ayarlamak için aşağıdaki tabloyu kullanın.

    Parametre Value
    İzin verilen hizmetler Blob
    İzin verilen kaynak türleri Nesne, Kapsayıcı
    Sağlanan izinler Okuma, Yazma, Silme, Listeleme, Oluşturma

    En az ayrıcalık için en iyi duruma getirmek için tek bir kapsayıcının SAS'sini almayı da seçebilirsiniz. Kimlik doğrulama hatalarını önlemek için kapsayıcının konu eşleme yapılandırmasındaki değerle eşleştiğinden table.tableName emin olun.

  2. SAS belirteci ile bir Kubernetes gizli dizisi oluşturun. Belirtecin başında olabilecek soru işaretini ? eklemeyin.

    kubectl create secret generic my-sas \
--from-literal=accessToken='sv=2022-11-02&ss=b&srt=c&sp=rwdlax&se=2023-07-22T05:47:40Z&st=2023-07-21T21:47:40Z&spr=https&sig=xDkwJUO....' \
-n azure-iot-operations

  3. Bağlayıcı için yapılandırma ve uç nokta ayarlarını tanımlayan bir DataLake Bağlan or kaynağı oluşturun. Örnek olarak sağlanan YAML'yi kullanabilirsiniz, ancak aşağıdaki alanları değiştirdiğinizden emin olun:

    • endpoint: AdLSv2 depolama hesabının biçimindeki https://example.blob.core.windows.netData Lake Depolama uç noktası. Azure portalında Uç Noktalar Data Lake Depolama Ayarlar Depolama > hesabı > altında uç noktayı > bulun.
    • accessTokenSecretName: SAS belirtecini içeren Kubernetes gizli dizisinin adı (my-sas önceki örnekten).
    apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnector
metadata:
  name: my-datalake-connector
  namespace: azure-iot-operations
spec:
  protocol: v5
  image:
    repository: mcr.microsoft.com/azureiotoperations/datalake
    tag: 0.4.0-preview
    pullPolicy: IfNotPresent
  instances: 2
  logLevel: "debug"
  databaseFormat: "delta"
  target:
    datalakeStorage:
      endpoint: "https://example.blob.core.windows.net"
      authentication:
        accessTokenSecretName: "my-sas"
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}

  4. Data Lake Depolama MQTT konusu ile Delta tablosu arasındaki eşlemeyi tanımlayan bir DataLake Bağlan orTopicMap kaynağı oluşturun. Örnek olarak sağlanan YAML'yi kullanabilirsiniz, ancak aşağıdaki alanları değiştirdiğinizden emin olun:

    • dataLakeConnectorRef: Daha önce oluşturduğunuz DataLake Bağlan or kaynağının adı.
    • clientId: MQTT istemciniz için benzersiz bir tanımlayıcı.
    • mqttSourceTopic: Verilerin gelmesini istediğiniz MQTT konusunun adı.
    • table.tableName: Data Lake Depolama eklemek istediğiniz kapsayıcının adı. SAS belirtecinin kapsamı hesap olarak belirlenmişse, eksikse kapsayıcı otomatik olarak oluşturulur.
    • table.schema: MQTT konusuna gönderdiğiniz JSON iletilerinin biçimi ve alanlarıyla eşleşmesi gereken Delta tablosunun şeması.

  5. Kullanarak kubectl apply -f datalake-connector.yamlKubernetes kümenize DataLake Bağlan or ve DataLake Bağlan orTopicMap kaynaklarını uygulayın.

  6. MQTT yayımcınızı kullanarak MQTT konusuna JSON iletileri göndermeye başlayın. Data Lake bağlayıcı örneği konuya abone olur ve iletileri Delta tablosuna alır.

  7. Azure portalını kullanarak Delta tablosunun oluşturulduğunu doğrulayın. Dosyalar istemci kimliği, bağlayıcı örneği adı, MQTT konusu ve zamana göre düzenlenir. Depolama hesabı >Kapsayıcılarınızda DataLake Bağlan orTopicMap içinde belirttiğiniz kapsayıcıyı açın. _delta_log olduğunu doğrulayın ve ayrıştırılmış dosyalar MQTT trafiğini gösterir. Yükün şemada gönderilen ve tanımlanan değerle eşleşip eşleşmedığını onaylamak için bir parque dosyası açın.

ADLSv2 kimlik doğrulaması için yönetilen kimliği kullanma

Yönetilen kimliği kullanmak için DataLake Bağlan or authenticationaltında tek yöntem olarak belirtin. IoT MQ Arc uzantısının asıl kimliğini bulmak için kullanınaz k8s-extension show, ardından yönetilen kimliğe depolama hesabına yazma izni veren Depolama Blob Veri Katkıda Bulunanı gibi bir rol atayın. Daha fazla bilgi edinmek için bkz . Microsoft Entra Id kullanarak bloblara erişimi yetkilendirme.

authentication:
  systemAssignedManagedIdentity:
    audience: https://my-account.blob.core.windows.net

Yönetilen kimlik kullanarak Azure Veri Gezgini'a veri gönderecek şekilde yapılandırma

Data Lake bağlayıcısını yönetilen kimliği kullanarak azure Veri Gezgini uç noktasına veri gönderecek şekilde yapılandırın.

  1. Tam azure Veri Gezgini kümesi de dahil olmak üzere önkoşullardaki adımların karşılandığından emin olun. "Ücretsiz küme" seçeneği çalışmıyor.

  2. Küme oluşturulduktan sonra verilerinizi depolamak için bir veritabanı oluşturun.

  3. Azure portalı aracılığıyla belirli veriler için bir tablo oluşturabilir ve sütunları el ile oluşturabilir veya sorgu sekmesindeki KQL'yi kullanabilirsiniz. Örneğin:

    .create table thermostat (
    externalAssetId: string,
    assetName: string,
    CurrentTemperature: real,
    Pressure: real,
    MqttTopic: string,
    Timestamp: datetime
)

Akış alımını etkinleştirme

Tablonuzda ve veritabanınızda akış alımını etkinleştirin. Sorgu sekmesinde aşağıdaki komutu çalıştırın ve <DATABASE_NAME> veritabanınızın adını yazın:

.alter database <DATABASE_NAME> policy streamingingestion enable

Yönetilen kimliği Azure Veri Gezgini kümesine ekleme

Bağlayıcının Azure Veri Gezgini kimlik doğrulaması yapabilmesi için yönetilen kimliği Azure Veri Gezgini kümesine eklemeniz gerekir.

  1. Azure portalında Arc bağlantılı Kubernetes kümesine gidin ve Ayarlar> Extensions'ı seçin. Uzantı listesinde IoT MQ uzantınızın adını arayın. Ad, ardından beş rastgele karakterle mq- başlar. Örneğin, mq-4jgjs. IoT MQ uzantısı adı, MQ yönetilen kimlik adıyla aynıdır.
  2. Azure Veri Gezgini veritabanınızda İzinler>Alma Ekle'yi> seçin. MQ yönetilen kimlik adını arayın ve ekleyin.

İzin ekleme hakkında daha fazla bilgi için bkz. Azure Veri Gezgini kümesi izinlerini yönetme.

Artık bağlayıcıyı dağıtmaya ve Azure Veri Gezgini'a veri göndermeye hazırsınız.

Örnek dağıtım dosyası

Azure Veri Gezgini bağlayıcısı için örnek dağıtım dosyası. ile TODO başlayan açıklamalar için yer tutucu ayarlarını bilgilerinizle değiştirmeniz gerekir.

apiVersion: mq.iotoperations.azure.com/v1beta1
  name: my-adx-connector
  namespace: azure-iot-operations
spec:
    repository: mcr.microsoft.com/azureiotoperations/datalake
    tag: 0.4.0-preview
    pullPolicy: Always
  databaseFormat: adx
  target:
      # TODO: insert the ADX cluster endpoint
      endpoint: https://<CLUSTER>.<REGION>.kusto.windows.net
      authentication:
        systemAssignedManagedIdentity:
          audience: https://api.kusto.windows.net
  localBrokerConnection:
    endpoint: aio-mq-dmqtt-frontend:8883
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: aio-ca-trust-bundle-test-only
    authentication:
      kubernetes: {}
---
apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnectorTopicMap
metadata:
  name: adx-topicmap
  namespace: azure-iot-operations
spec:
  mapping:
    allowedLatencySecs: 1
    messagePayloadType: json
    maxMessagesPerBatch: 10
    clientId: id
    mqttSourceTopic: azure-iot-operations/data/thermostat
    qos: 1
    table:
      # TODO: add DB and table name
      tablePath: <DATABASE_NAME>
      tableName: <TABLE_NAME>
      schema:
      - name: externalAssetId
        format: utf8
        optional: false
        mapping: $property.externalAssetId
      - name: assetName
        format: utf8
        optional: false
        mapping: DataSetWriterName
      - name: CurrentTemperature
        format: float32
        optional: false
        mapping: Payload.temperature.Value
      - name: Pressure
        format: float32
        optional: true
        mapping: "Payload.Tag 10.Value"
      - name: MqttTopic
        format: utf8
        optional: false
        mapping: $topic
      - name: Timestamp
        format: timestamp
        optional: false
        mapping: $received_time

Bu örnek, aşağıdaki gibi JSON biçiminde iletiler içeren konu başlığındaki azure-iot-operations/data/thermostat verileri kabul eder:

{
  "SequenceNumber": 4697,
  "Timestamp": "2024-04-02T22:36:03.1827681Z",
  "DataSetWriterName": "thermostat",
  "MessageType": "ua-deltaframe",
  "Payload": {
    "temperature": {
      "SourceTimestamp": "2024-04-02T22:36:02.6949717Z",
      "Value": 5506
    },
    "Tag 10": {
      "SourceTimestamp": "2024-04-02T22:36:02.6949888Z",
      "Value": 5506
    }
  }
}

DataLake Bağlan or

DataLake Bağlan or, data lake bağlayıcı örneğinin yapılandırmasını ve özelliklerini tanımlayan bir Kubernetes özel kaynağıdır. Data Lake bağlayıcısı, MQTT konu başlıklarından verileri Data Lake Depolama hesabındaki Delta tablolarına alır.

DataLake Bağlan or kaynağının belirtim alanı aşağıdaki alt alanları içerir:

  • protocol: MQTT sürümü. veya v3olabilirv5.
  • image: Görüntü alanı, data lake bağlayıcı modülünün kapsayıcı görüntüsünü belirtir. Aşağıdaki alt alanlara sahiptir:
    • repository: Görüntünün depolandığı kapsayıcı kayıt defterinin ve deponun adı.
    • tag: Kullanılacak görüntünün etiketi.
    • pullPolicy: Görüntünün çekme ilkesi. , veya NeverolabilirAlwaysIfNotPresent.
  • instances: Data Lake bağlayıcısının çalıştırılacak çoğaltma sayısı.
  • logLevel: Data Lake Bağlayıcı modülü için günlük düzeyi. , , debug, info, warnerrorveya fatalolabilirtrace.
  • databaseFormat: Data Lake Depolama alınacak verilerin biçimi. veya parquetolabilirdelta.
  • target: Hedef alan, veri alımının hedefini belirtir. , , fabricOneLakeadxveya localStorageolabilirdatalakeStorage.
    • datalakeStorage: ADLSv2 hesabının yapılandırmasını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
      • endpoint: Data Lake Depolama hesap uç noktasının URL'si. Sondaki eğik çizgiyi /eklemeyin.
      • authentication: Kimlik doğrulama alanı, Data Lake Depolama hesabına erişmek için türü ve kimlik bilgilerini belirtir. Aşağıdakilerden biri olabilir.
        • accessTokenSecretName: Data Lake Depolama hesabı için paylaşılan erişim belirteci kimlik doğrulamasını kullanmaya yönelik Kubernetes gizli dizisinin adı. Türü accessTokenise bu alan gereklidir.
        • systemAssignedManagedIdentity: Kimlik doğrulaması için sistem tarafından yönetilen kimlik kullanmak için. Bir alt alanı var
          • audience: Hesap düzeyinde veya https://storage.azure.com herhangi bir depolama hesabı için kapsamı belirlenmiş yönetilen kimlik belirteci hedef kitlesi için biçiminde https://<my-account-name>.blob.core.windows.net bir dize.
    • fabricOneLake: Microsoft Fabric OneLake'in yapılandırmasını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
      • endpoint: Microsoft Fabric OneLake uç noktasının URL'si. Bunun nedeni genellikle https://onelake.dfs.fabric.microsoft.com OneLake genel uç noktası olmasıdır. Bölgesel uç nokta kullanıyorsanız, biçimindedir https://<region>-onelake.dfs.fabric.microsoft.com. Sondaki eğik çizgiyi /eklemeyin. Daha fazla bilgi edinmek için bkz. Microsoft OneLake'e Bağlan.
      • names: Çalışma alanının ve göl evi adlarını belirtir. Bu alanı veya guidskullanın. İkisini birden kullanmayın. Aşağıdaki alt alanlara sahiptir:
        • workspaceName: Çalışma alanının adı.
        • lakehouseName: Göl evi adı.
      • guids: Çalışma alanının ve lakehouse'un GUID'lerini belirtir. Bu alanı veya nameskullanın. İkisini birden kullanmayın. Aşağıdaki alt alanlara sahiptir:
        • workspaceGuid: Çalışma alanının GUID değeri.
        • lakehouseGuid: Göl evi GUID'i.
      • fabricPath: Doku çalışma alanında verilerin konumu. veya filesolabilirtables. Ise tablesveriler Doku OneLake'te tablo olarak depolanır. Bu filesise, veriler Doku OneLake'te dosya olarak depolanır. bu ise files, databaseFormat olmalıdır parquet.
      • authentication: Kimlik doğrulama alanı, Microsoft Fabric OneLake'e erişmek için türü ve kimlik bilgilerini belirtir. Sadece şimdilik olabilir systemAssignedManagedIdentity . Bir alt alanı vardır:
      • systemAssignedManagedIdentity: Kimlik doğrulaması için sistem tarafından yönetilen kimlik kullanmak için. Bir alt alanı var
        • audience: Yönetilen kimlik belirteci hedef kitlesi için bir dize ve olmalıdır https://storage.azure.com.
    • adx: Azure Veri Gezgini veritabanının yapılandırmasını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
      • endpoint: gibi https://<CLUSTER>.<REGION>.kusto.windows.netAzure Veri Gezgini kümesi uç noktasının URL'si. Sondaki eğik çizgiyi /eklemeyin.
      • authentication: Kimlik doğrulama alanı, Azure Veri Gezgini kümesine erişmek için türü ve kimlik bilgilerini belirtir. Sadece şimdilik olabilir systemAssignedManagedIdentity . Bir alt alanı vardır:
        • systemAssignedManagedIdentity: Kimlik doğrulaması için sistem tarafından yönetilen kimlik kullanmak için. Bir alt alanı var
          • audience: Yönetilen kimlik belirteci hedef kitlesi için bir dize ve olmalıdır https://api.kusto.windows.net.
    • localStorage: Yerel depolama hesabının yapılandırmasını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
      • volumeName: Bağlayıcı podlarının her birine bağlanan birimin adı.
  • localBrokerConnection: IoT MQ MQ MQTT aracısına varsayılan bağlantı yapılandırmasını geçersiz kılmak için kullanılır. Bkz. Yerel aracı bağlantısını yönetme.

DataLake Bağlan orTopicMap

DataLake Bağlan orTopicMap, Bir MQTT konusu ile Data Lake Depolama hesabındaki Delta tablosu arasındaki eşlemeyi tanımlayan bir Kubernetes özel kaynağıdır. DataLake Bağlan orTopicMap kaynağı, aynı uç cihazda çalışan ve MQTT konusundan Delta tablosuna veri alan dataLake Bağlan or kaynağına başvurur.

DataLake Bağlan orTopicMap kaynağının belirtim alanı aşağıdaki alt alanları içerir:

  • dataLakeConnectorRef: Bu konu haritasının ait olduğu DataLake Bağlan or kaynağının adı.
  • mapping: Eşleme alanı, MQTT konusunun ve Delta tablosunun ayrıntılarını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
    • allowedLatencySecs: MQTT konusundan ileti alma ve Delta tablosuna alma arasında saniye cinsinden en yüksek gecikme süresi. Bu alan gereklidir.
    • clientId: Konuya abone olan MQTT istemcisinin benzersiz tanımlayıcısı.
    • maxMessagesPerBatch: Delta tablosuna tek bir toplu işlemde alınacak ileti sayısı üst sınırı. Geçici bir kısıtlama nedeniyle, 1 olarak ayarlandıysa qos bu değer 16'dan küçük olmalıdır. Bu alan gereklidir.
    • messagePayloadType: MQTT konusuna gönderilen yük türü. Bunlardan biri jsonavro veya (henüz desteklenmiyor) olabilir.
    • mqttSourceTopic: Abone olunacak MQTT konularının adı. MQTT konu joker karakter gösterimini destekler.
    • qos: MQTT konusuna abone olmak için hizmet düzeyi kalitesi. 0 veya 1 olabilir.
    • table: Tablo alanı, Data Lake Depolama hesabındaki Delta tablosunun yapılandırmasını ve özelliklerini belirtir. Aşağıdaki alt alanlara sahiptir:
      • tableName: Data Lake Depolama hesabında oluşturulacak veya eklenecek Delta tablosunun adı. Bu alan, Azure Data Lake Storage 2. Nesil ile kullanıldığında kapsayıcı adı olarak da bilinir. Uzunluğu 256 karaktere kadar olan herhangi bir küçük harf İngilizce harfi ve alt çubuk _içerebilir. Tire - veya boşluk karakterine izin verilmez.
      • tablePath: Tür bağlayıcısı kullanılırken adx Azure Veri Gezgini veritabanının adı.
      • schema: İleti yükünün biçimi ve alanlarıyla eşleşmesi gereken Delta tablosunun şeması. Her biri aşağıdaki alt alanlara sahip bir nesne dizisidir:
        • name: Delta tablosundaki sütunun adı.
        • format: Delta tablosundaki sütunun veri türü. , , , , , int32, int64, , uInt8, uInt16, , uInt32, , uInt64, float16, float32timestampbinaryfloat64date32veya .booleanutf8int16int8 gibi uInt8imzasız türler tam olarak desteklenmez ve burada belirtilirse imzalı türler olarak kabul edilir.
        • optional: Sütunun isteğe bağlı mı yoksa gerekli mi olduğunu gösteren boole değeri. Bu alan isteğe bağlıdır ve varsayılan olarak false olarak ayarlanır.
        • mapping: MQTT ileti yükünden sütunun değerini ayıklamayı tanımlayan JSON yol ifadesi. Yerleşik , , $topic$propertyve $received_time eşlemeleri$client_id, MQTT ileti gövdesinde JSON'u zenginleştirmek için sütun olarak kullanılabilir. Bu alan gereklidir. MQTT kullanıcı özellikleri için $property kullanın. Örneğin, $property.assetId MQTT iletisindeki assetId özelliğinin değerini temsil eder.

DataLake Bağlan orTopicMap kaynağı örneği aşağıda verilmiştir:

apiVersion: mq.iotoperations.azure.com/v1beta1
kind: DataLakeConnectorTopicMap
metadata:
  name: datalake-topicmap
  namespace: azure-iot-operations
spec:
  dataLakeConnectorRef: my-datalake-connector
  mapping:
    allowedLatencySecs: 1
    messagePayloadType: json
    maxMessagesPerBatch: 10
    clientId: id
    mqttSourceTopic: azure-iot-operations/data/thermostat
    qos: 1
    table:
      tableName: thermostat
      schema:
      - name: externalAssetId
        format: utf8
        optional: false
        mapping: $property.externalAssetId
      - name: assetName
        format: utf8
        optional: false
        mapping: DataSetWriterName
      - name: CurrentTemperature
        format: float32
        optional: false
        mapping: Payload.temperature.Value
      - name: Pressure
        format: float32
        optional: true
        mapping: "Payload.Tag 10.Value"
      - name: Timestamp
        format: timestamp
        optional: false
        mapping: $received_time

gibi "{\"SequenceNumber\": 4697, \"Timestamp\": \"2024-04-02T22:36:03.1827681Z\", \"DataSetWriterName\": \"thermostat-de\", \"MessageType\": \"ua-deltaframe\", \"Payload\": {\"temperature\": {\"SourceTimestamp\": \"2024-04-02T22:36:02.6949717Z\", \"Value\": 5506}, \"Tag 10\": {\"SourceTimestamp\": \"2024-04-02T22:36:02.6949888Z\", \"Value\": 5506}}}" dizeli JSON desteklenmez ve bağlayıcının null değer hatası bulduğu bir dönüştürücü oluşturmasına neden olur.

Bu şemayla çalışan konu için azure-iot-operations/data/thermostat örnek bir ileti:

{
  "SequenceNumber": 4697,
  "Timestamp": "2024-04-02T22:36:03.1827681Z",
  "DataSetWriterName": "thermostat",
  "MessageType": "ua-deltaframe",
  "Payload": {
    "temperature": {
      "SourceTimestamp": "2024-04-02T22:36:02.6949717Z",
      "Value": 5506
    },
    "Tag 10": {
      "SourceTimestamp": "2024-04-02T22:36:02.6949888Z",
      "Value": 5506
    }
  }
}

Hangi eşlemeler:

externalAssetId assetName CurrentTemperature Basınç mqttTopic timestamp
xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx thermostat-de 5506 5506 Dlc 2024-04-02T22:36:03.1827681Z

Önemli

Veri şeması güncelleştirilirse, örneğin bir veri türü değiştirilirse veya bir ad değiştirilirse, gelen verilerin dönüşümü çalışmayı durdurabilir. Şema değişikliği olursa veri tablosu adını değiştirmeniz gerekir.

Delta veya parke

Hem delta hem de parquet biçimleri desteklenir.

Yerel aracı bağlantısını yönetme

MQTT köprüsü gibi data lake bağlayıcısı da IoT MQ MQ MQTT aracısı için bir istemci işlevi görür. IoT MQ MQTT aracınızın dinleyici bağlantı noktasını veya kimlik doğrulamasını özelleştirdiyseniz data lake bağlayıcısı için yerel MQTT bağlantı yapılandırmasını da geçersiz kılın. Daha fazla bilgi edinmek için bkz . MQTT köprüsü yerel aracı bağlantısı.

İlgili içerik

Azure IoT MQ Preview kullanarak MQTT iletilerini yayımlama ve abone olma