Konfigurace koncových bodů toku dat Azure Event Hubs a Kafka

Článek
11/20/2024

Důležité

Tato stránka obsahuje pokyny ke správě komponent operací Azure IoT pomocí manifestů nasazení Kubernetes, které jsou ve verzi Preview. Tato funkce je poskytována s několika omezeními a neměla by se používat pro produkční úlohy.

Právní podmínky, které platí pro funkce Azure, které jsou ve verzi beta, verzi Preview nebo které zatím nejsou veřejně dostupné, najdete v Dodatečných podmínkách použití pro Microsoft Azure verze Preview.

Pokud chcete nastavit obousměrnou komunikaci mezi operacemi Azure IoT a zprostředkovateli Apache Kafka, můžete nakonfigurovat koncový bod toku dat. Tato konfigurace umožňuje zadat koncový bod, protokol TLS (Transport Layer Security), ověřování a další nastavení.

Požadavky

Instance operací Azure IoT

Azure Event Hubs

Služba Azure Event Hubs je kompatibilní s protokolem Kafka a dá se používat s toky dat s určitými omezeními.

Vytvoření oboru názvů a centra událostí Služby Azure Event Hubs

Nejprve vytvořte obor názvů Azure Event Hubs s podporou Kafka.

Dále vytvořte centrum událostí v oboru názvů. Každé jednotlivé centrum událostí odpovídá tématu Kafka. Ve stejném oboru názvů můžete vytvořit více center událostí, které představují více témat Kafka.

Přiřazení oprávnění ke spravované identitě

Pokud chcete nakonfigurovat koncový bod toku dat pro službu Azure Event Hubs, doporučujeme použít spravovanou identitu přiřazenou uživatelem nebo systémem. Tento přístup je zabezpečený a eliminuje potřebu správy přihlašovacích údajů ručně.

Po vytvoření oboru názvů a centra událostí služby Azure Event Hubs musíte přiřadit roli spravované identitě azure IoT Operations, která uděluje oprávnění k odesílání nebo přijímání zpráv do centra událostí.

Pokud používáte spravovanou identitu přiřazenou systémem, přejděte na webu Azure Portal do instance operací Azure IoT a vyberte Přehled. Zkopírujte název rozšíření uvedené za rozšířením Azure IoT Operations Arc. Například azure-iot-operations-xxxx7. Spravovanou identitu přiřazenou systémem najdete pomocí stejného názvu rozšíření Azure IoT Operations Arc.

Pak přejděte do řízení přístupu k oboru názvů >Služby Event Hubs (IAM)>Přidat přiřazení role.

Na kartě Role vyberte odpovídající roli, například Azure Event Hubs Data Sender nebo Azure Event Hubs Data Receiver. Díky tomu má spravovaná identita potřebná oprávnění k odesílání nebo přijímání zpráv pro všechna centra událostí v oboru názvů. Další informace najdete v tématu Ověření aplikace pomocí ID Microsoft Entra pro přístup k prostředkům služby Event Hubs.
Na kartě Členové:
1. Pokud používáte spravovanou identitu přiřazenou systémem, pro přiřazení přístupu, vyberte možnost Uživatel, skupina nebo instanční objekt , pak vyberte a vyberte členy a vyhledejte název rozšíření Azure IoT Operations Arc.
2. Pokud používáte spravovanou identitu přiřazenou uživatelem, jako možnost Přiřadit přístup, vyberte možnost Spravovaná identita a pak vyberte a vyberte členy a vyhledejte spravovanou identitu přiřazenou uživatelem nastavenou pro cloudová připojení.

Vytvoření koncového bodu toku dat pro Azure Event Hubs

Jakmile je obor názvů a centrum událostí Azure Event Hubs nakonfigurované, můžete vytvořit koncový bod toku dat pro obor názvů Azure Event Hubs s podporou Kafka.

V provozním prostředí vyberte kartu Koncové body toku dat.
V části Vytvořit nový koncový bod toku dat vyberte Azure Event Hubs>New.

Zadejte následující nastavení pro koncový bod:

Nastavení	Description
Name	Název koncového bodu toku dat.
Hostitelský počítač	Název hostitele zprostředkovatele Kafka ve formátu `<NAMESPACE>.servicebus.windows.net:9093`. Zahrňte číslo `9093` portu do nastavení hostitele pro službu Event Hubs.
Metoda ověřování	Metoda použitá k ověřování. Doporučujeme zvolit spravovanou identitu přiřazenou systémem nebo spravovanou identitu přiřazenou uživatelem.

Vyberte Použít pro zřízení koncového bodu.

Vytvořte soubor Bicep .bicep s následujícím obsahem.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param endpointName string = '<ENDPOINT_NAME>'
param hostName string = '<NAMESPACE>.servicebus.windows.net:9093'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}
resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: endpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    kafkaSettings: {
      host: hostName
      authentication: {
        // See available authentication methods section for method types
        // method: <METHOD_TYPE>
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

Pak nasaďte pomocí Azure CLI.

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

Vytvořte soubor manifestu .yaml Kubernetes s následujícím obsahem.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT_NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <NAMESPACE>.servicebus.windows.net:9093
    authentication:
      # See available authentication methods section for method types
      # method: <METHOD_TYPE>
    tls:
      mode: Enabled

Pak použijte soubor manifestu na cluster Kubernetes.

kubectl apply -f <FILE>.yaml

Poznámka:

Téma Kafka nebo samostatné centrum událostí se později nakonfiguruje při vytváření toku dat. Téma Kafka je cílem zpráv toku dat.

Použití připojovací řetězec pro ověřování ve službě Event Hubs

Důležité

Pokud chcete ke správě tajných kódů použít portál provozního prostředí, musí být operace Azure IoT nejprve povoleny se zabezpečeným nastavením konfigurací služby Azure Key Vault a povolením identit úloh. Další informace najdete v tématu Povolení nastavení zabezpečení v nasazení operací Azure IoT.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Základní a pak zvolte SASL metody>ověřování.

Zadejte následující nastavení pro koncový bod:

Nastavení	Popis
Typ SASL	Zvolte `Plain`.
Synchronizovaný název tajného kódu	Zadejte název tajného kódu Kubernetes, který obsahuje připojovací řetězec.
Referenční informace k uživatelskému jménu nebo tajný klíč tokenu	Odkaz na tajný klíč uživatelského jména nebo tokenu použitého pro ověřování SASL. Vyberte ho ze seznamu služby Key Vault nebo vytvořte novou. Hodnota musí být `$ConnectionString`.
Odkaz na heslo tajného kódu tokenu	Odkaz na tajný klíč hesla nebo tokenu použitého pro ověřování SASL. Vyberte ho ze seznamu služby Key Vault nebo vytvořte novou. Hodnota musí být ve formátu `Endpoint=sb://<NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>`.

Po výběru možnosti Přidat odkaz, pokud vyberete Vytvořit nový, zadejte následující nastavení:

Nastavení	Popis
Název tajného klíče	Název tajného kódu ve službě Azure Key Vault. Vyberte název, který si můžete snadno zapamatovat, abyste ho později ze seznamu vybrali.
Hodnota tajného klíče	Jako uživatelské jméno zadejte `$ConnectionString`. Jako heslo zadejte připojovací řetězec ve formátu `Endpoint=sb://<NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>`.
Nastavení data aktivace	Pokud je zapnuté, datum, kdy se tajný klíč aktivuje.
Nastavení data vypršení platnosti	Pokud je tato možnost zapnutá, datum vypršení platnosti tajného kódu.

Další informace o tajných kódech najdete v tématu Vytváření a správa tajných kódů v operacích Azure IoT.

kafkaSettings: {
  authentication: {
    method: 'Sasl'
    saslSettings: {
      saslType: 'Plain'
      secretRef: '<SECRET_NAME>'
    }
  }
  tls: {
    mode: 'Enabled'
  }
}

Pokud chcete použít připojovací řetězec pro ověřování ve službě Event Hubs, použijte metodu ověřování SASL a nakonfigurujte typ SASL jako prostý a nakonfigurujte název tajného klíče, který obsahuje připojovací řetězec.

Nejprve vytvořte tajný klíč Kubernetes, který obsahuje připojovací řetězec. Tajný klíč musí být ve stejném oboru názvů jako koncový bod toku dat Kafka. Tajný klíč musí mít jako páry klíč-hodnota uživatelské jméno i heslo. Příklad:

kubectl create secret generic <SECRET_NAME> -n azure-iot-operations \
  --from-literal=username='$ConnectionString' \
  --from-literal=password='Endpoint=sb://<NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>'

Tip

Rozsah připojovací řetězec oboru názvů (na rozdíl od jednotlivých center událostí) umožňuje toku dat odesílat a přijímat zprávy z několika různých center událostí a témat Kafka.

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain
      secretRef: <SECRET_NAME>
  tls:
    mode: Enabled

Omezení

Azure Event Hubs nepodporuje všechny typy komprese, které Kafka podporuje. V současné době se ve službě Azure Event Hubs premium a vyhrazených úrovních podporuje pouze komprese GZIP. Použití jiných typů komprese může vést k chybám.

Vlastní zprostředkovatelé Kafka

Pokud chcete nakonfigurovat koncový bod toku dat pro zprostředkovatele Kafka mimo centrum událostí, podle potřeby nastavte hostitele, tls, ověřování a další nastavení.

V provozním prostředí vyberte kartu Koncové body toku dat.
V části Vytvořit nový koncový bod toku dat vyberte Vlastní zprostředkovatel Kafka Nový>.

Zadejte následující nastavení pro koncový bod:

Nastavení	Description
Name	Název koncového bodu toku dat.
Hostitelský počítač	Název hostitele zprostředkovatele Kafka ve formátu `<Kafka-broker-host>:xxxx`. Do nastavení hostitele zahrňte číslo portu.
Metoda ověřování	Metoda použitá k ověřování. Zvolte SASL.
Typ SASL	Typ ověřování SASL. Zvolte Plain, ScramSha256 nebo ScramSha512. Vyžaduje se, pokud používáte SASL.
Synchronizovaný název tajného kódu	Název tajného klíče. Vyžaduje se, pokud používáte SASL.
Odkaz na uživatelské jméno tajného kódu tokenu	Odkaz na uživatelské jméno v tajném kódu tokenu SASL. Vyžaduje se, pokud používáte SASL.

Vyberte Použít pro zřízení koncového bodu.

Poznámka:

V současné době provozní prostředí nepodporuje použití koncového bodu toku dat Kafka jako zdroje. Tok dat se zdrojovým koncovým bodem toku dat Kafka můžete vytvořit pomocí Kubernetes nebo Bicep.

resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
 parent: aioInstance
  name: '<ENDPOINT NAME>'
  extendedLocation: {
    name: '<CUSTOM_LOCATION_NAME>'
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    host: '<KAFKA-HOST>:<PORT>'
    kafkaSettings: {
      authentication: {
        method: 'Sasl'
        saslSettings: {
            saslType: '<TYPE>'
            secretRef: '<SECRET_NAME>'
        }
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <KAFKA-HOST>:<PORT>
    authentication:
      method: Sasl
      saslSettings:
        saslType: <TYPE>
        secretRef: <SECRET_NAME>
    tls:
      mode: Enabled

Pokud chcete přizpůsobit nastavení koncového bodu, další informace najdete v následujících částech.

Dostupné metody ověřování

Pro koncové body toku dat zprostředkovatele Kafka jsou k dispozici následující metody ověřování.

Spravovaná identita přiřazená systémem

Před konfigurací koncového bodu toku dat přiřaďte roli spravované identitě operací Azure IoT, která uděluje oprávnění pro připojení ke zprostředkovateli Kafka:

Na webu Azure Portal přejděte do instance operace Azure IoT a vyberte Přehled.
Zkopírujte název rozšíření uvedené za rozšířením Azure IoT Operations Arc. Například azure-iot-operations-xxxx7.
Přejděte do cloudového prostředku, který potřebujete udělit oprávnění. Přejděte například do řízení přístupu k oboru názvů >služby Event Hubs (IAM)>Přidat přiřazení role.
Na kartě Role vyberte příslušnou roli.
Na kartě Členové vyberte možnost Přiřadit přístup, vyberte Možnost Uživatel, skupina nebo instanční objekt a pak vyberte a vyberte členy a vyhledejte spravovanou identitu operací Azure IoT. Například azure-iot-operations-xxxx7.

Pak nakonfigurujte koncový bod toku dat s nastavením spravované identity přiřazené systémem.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Základní a pak zvolte spravovanou identitu přiřazenou systémem ověřování>.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {}
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      {}

Tato konfigurace vytvoří spravovanou identitu s výchozí cílovou skupinou, která je stejná jako hodnota hostitele oboru názvů Event Hubs ve formě https://<NAMESPACE>.servicebus.windows.net. Pokud ale potřebujete přepsat výchozí cílovou skupinu, můžete pole nastavit audience na požadovanou hodnotu.

Nepodporuje se v provozním prostředí.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {
        audience: '<YOUR_AUDIENCE_OVERRIDE_VALUE>'
    }
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      audience: <YOUR_AUDIENCE_OVERRIDE_VALUE>

Spravovaná identita přiřazená uživatelem

Pokud chcete pro ověřování použít spravovanou identitu přiřazenou uživatelem, musíte nejprve nasadit operace Azure IoT s povoleným zabezpečeným nastavením. Pak musíte nastavit spravovanou identitu přiřazenou uživatelem pro cloudová připojení. Další informace najdete v tématu Povolení nastavení zabezpečení v nasazení operací Azure IoT.

Před konfigurací koncového bodu toku dat přiřaďte spravované identitě přiřazené uživatelem roli, která uděluje oprávnění pro připojení ke zprostředkovateli Kafka:

Na webu Azure Portal přejděte do cloudového prostředku, který potřebujete udělit oprávnění. Přejděte například do řízení přístupu k oboru názvů >Event Gridu (IAM)>Přidat přiřazení role.
Na kartě Role vyberte příslušnou roli.
Na kartě Členové vyberte možnost Přiřadit přístup, vyberte možnost Spravovaná identita a pak vyberte a vyberte členy a vyhledejte spravovanou identitu přiřazenou uživatelem.

Pak nakonfigurujte koncový bod toku dat s nastavením spravované identity přiřazené uživatelem.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Základní a pak zvolte Spravovaná identita přiřazená metodou>ověřování.

kafkaSettings: {
  authentication: {
    method: 'UserAssignedManagedIdentity'
    UserAssignedManagedIdentitySettings: {
      clientId: '<CLIENT_ID>'
      tenantId: '<TENANT_ID>'
      // Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      // Matching the Event Hub namespace you configured as host
      // scope: 'https://<SCOPE_URL>'
    }
  }
  ...
}

kafkaSettings:
  authentication:
    method: UserAssignedManagedIdentity
    userAssignedManagedIdentitySettings:
      clientId: <CLIENT_ID>
      tenantId: <TENANT_ID>
      # Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      # Matching the Event Hub namespace you configured as host
      # scope: https://<SCOPE_URL>

V této části je rozsah cílovou skupinou spravované identity. Výchozí hodnota je stejná jako hodnota hostitele oboru názvů Event Hubs ve formě https://<NAMESPACE>.servicebus.windows.net. Pokud ale potřebujete přepsat výchozí cílovou skupinu, můžete pole oboru nastavit na požadovanou hodnotu pomocí Bicep nebo Kubernetes.

SASL

Pokud chcete k ověřování použít SASL, zadejte metodu ověřování SASL a nakonfigurujte typ SASL a odkaz na tajný klíč s názvem tajného klíče, který obsahuje token SASL.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Základní a pak zvolte SASL metody>ověřování.

Zadejte následující nastavení pro koncový bod:

Nastavení	Popis
Typ SASL	Typ ověřování SASL, který se má použít. Podporované typy jsou `Plain`, `ScramSha256`a `ScramSha512`.
Synchronizovaný název tajného kódu	Název tajného kódu Kubernetes, který obsahuje token SASL.
Referenční informace k uživatelskému jménu nebo tajný klíč tokenu	Odkaz na tajný klíč uživatelského jména nebo tokenu použitého pro ověřování SASL.
Odkaz na heslo tajného kódu tokenu	Odkaz na tajný klíč hesla nebo tokenu použitého pro ověřování SASL.

kafkaSettings: {
  authentication: {
    method: 'Sasl' // Or ScramSha256, ScramSha512
    saslSettings: {
      saslType: 'Plain' // Or ScramSha256, ScramSha512
      secretRef: '<SECRET_NAME>'
    }
  }
}

kubectl create secret generic sasl-secret -n azure-iot-operations \
  --from-literal=token='<YOUR_SASL_TOKEN>'

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain # Or ScramSha256, ScramSha512
      secretRef: <SECRET_NAME>

Podporované typy SASL jsou:

Plain
ScramSha256
ScramSha512

Tajný klíč musí být ve stejném oboru názvů jako koncový bod toku dat Kafka. Tajný klíč musí mít token SASL jako pár klíč-hodnota.

Anonymní

Pokud chcete použít anonymní ověřování, aktualizujte část ověřování nastavení Kafka tak, aby používala anonymní metodu.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Základní a pak zvolte Metodu>ověřování None.

kafkaSettings: {
  authentication: {
    method: 'Anonymous'
  }
}

kafkaSettings:
  authentication:
    method: Anonymous
    anonymousSettings:
      {}

Rozšířené nastavení

Můžete nastavit upřesňující nastavení pro koncový bod toku dat Kafka, jako je TLS, důvěryhodný certifikát CA, nastavení zasílání zpráv Kafka, dávkování a CloudEvents. Tato nastavení můžete nastavit na kartě Portál Advanced Portal koncového bodu toku dat nebo v rámci prostředku koncového bodu toku dat.

V provozním prostředí vyberte kartu Upřesnit pro koncový bod toku dat.

Snímek obrazovky s provozním prostředím k nastavení upřesňujícího nastavení koncového bodu toku dat Kafka

V části kafkaSettingsMůžete nakonfigurovat další nastavení pro koncový bod Kafka.

// See sections below for more details
kafkaSettings: {
  consumerGroupId: '<ID>'
  compression: 'Gzip'
  copyMqttProperties: 'Disabled'
  kafkaAcks: 'All'
  partitionStrategy: 'Default'
  tls: {
    mode: 'Enabled'
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
  batching: {
    mode: 'Enabled'
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

V části kafkaSettingsMůžete nakonfigurovat další nastavení pro koncový bod Kafka.

# See sections below for more details
kafkaSettings:
  consumerGroupId: <ID>
  compression: Gzip
  copyMqttProperties: Disabled
  kafkaAcks: All
  partitionStrategy: Default
  tls:
    mode: Enabled
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>
  batching:
    mode: Enabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Nastavení protokolu TLS

Režim TLS

Pokud chcete povolit nebo zakázat protokol TLS pro koncový bod Kafka, aktualizujte mode nastavení v nastavení protokolu TLS.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak použijte zaškrtávací políčko vedle povoleného režimu TLS.

kafkaSettings: {
  tls: {
    mode: 'Enabled' // Or Disabled
  }
}

kafkaSettings:
  tls:
    mode: Enabled # Or Disabled

Režim TLS lze nastavit na Enabled nebo Disabled. Pokud je režim nastavený na Enabled, tok dat používá zabezpečené připojení ke zprostředkovateli Kafka. Pokud je režim nastavený na Disabled, použije tok dat nezabezpečené připojení ke zprostředkovateli Kafka.

Certifikát důvěryhodné certifikační autority

Nakonfigurujte certifikát důvěryhodné certifikační autority pro koncový bod Kafka tak, aby vytvořil zabezpečené připojení ke zprostředkovateli Kafka. Toto nastavení je důležité, pokud zprostředkovatel Kafka používá certifikát podepsaný svým držitelem nebo certifikát podepsaný vlastní certifikační autoritou, která není ve výchozím nastavení důvěryhodná.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole mapování konfigurace certifikátu důvěryhodné certifikační autority zadejte objekt ConfigMap obsahující certifikát důvěryhodné certifikační autority.

kafkaSettings: {
  tls: {
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
}

kafkaSettings:
  tls:
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>

Tato mapa configmap by měla obsahovat certifikát certifikační autority ve formátu PEM. Objekt ConfigMap musí být ve stejném oboru názvů jako prostředek toku dat Kafka. Příklad:

kubectl create configmap client-ca-configmap --from-file root_ca.crt -n azure-iot-operations

Tip

Při připojování ke službě Azure Event Hubs se certifikát certifikační autority nevyžaduje, protože služba Event Hubs používá certifikát podepsaný veřejnou certifikační autoritou, která je ve výchozím nastavení důvěryhodná.

ID skupiny příjemců

ID skupiny příjemců slouží k identifikaci skupiny příjemců, kterou tok dat používá ke čtení zpráv z tématu Kafka. ID skupiny příjemců musí být jedinečné v rámci zprostředkovatele Kafka.

Důležité

Když se koncový bod Kafka použije jako zdroj, je potřeba ID skupiny příjemců. Jinak tok dat nemůže číst zprávy z tématu Kafka a zobrazí se chyba "Zdrojové koncové body typu Kafka musí mít definovaný consumerGroupId".

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole ID skupiny příjemce zadejte ID skupiny příjemců.

kafkaSettings: {
  consumerGroupId: '<ID>'
}

spec:
  kafkaSettings:
    consumerGroupId: <ID>

Toto nastavení se projeví jenom v případě, že se koncový bod používá jako zdroj (to znamená, že tok dat je příjemcem).

Komprese

Pole komprese umožňuje kompresi zpráv odeslaných do témat Kafka. Komprese pomáhá snížit šířku pásma sítě a prostor úložiště vyžadovaný pro přenos dat. Komprese ale také zvyšuje režii a latenci procesu. Podporované typy komprese jsou uvedeny v následující tabulce.

Hodnota	Popis
`None`	Nepoužívá se žádná komprese ani dávkování. Žádná není výchozí hodnota, pokud není zadána žádná komprese.
`Gzip`	Používá se komprese GZIP a dávkování. GZIP je algoritmus komprese pro obecné účely, který nabízí dobrou rovnováhu mezi poměrem komprese a rychlostí. V současné době se ve službě Azure Event Hubs premium a vyhrazených úrovních podporuje pouze komprese GZIP.
`Snappy`	Použije se komprese přichycení a dávkování. Snappy je algoritmus rychlé komprese, který nabízí střední poměr komprese a rychlost. Azure Event Hubs nepodporuje tento režim komprese.
`Lz4`	Používá se komprese LZ4 a dávkování. LZ4 je algoritmus rychlé komprese, který nabízí nízký poměr komprese a vysokou rychlost. Azure Event Hubs nepodporuje tento režim komprese.

Konfigurace komprese:

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole Komprese zadejte typ komprese.

kafkaSettings: {
  compression: 'Gzip' // Or Snappy, Lz4
}

kafkaSettings:
  compression: Gzip # Or Snappy, Lz4

Toto nastavení se projeví jenom v případě, že se koncový bod použije jako cíl, ve kterém je tok dat producentem.

Dávkování

Kromě komprese můžete také nakonfigurovat dávkování zpráv před jejich odesláním do témat Kafka. Dávkování umožňuje seskupit více zpráv a zkomprimovat je jako jednu jednotku, což může zlepšit efektivitu komprese a snížit režii sítě.

Pole	Popis	Povinní účastníci
`mode`	Může být `Enabled` nebo `Disabled`. Výchozí hodnota je `Enabled` , že Kafka nemá představu o nezvěsených zprávách. Pokud je nastavená hodnota `Disabled`, dávkování se minimalizuje, aby se při každém vytvoření dávky s jednou zprávou vytvořila.	No
`latencyMs`	Maximální časový interval v milisekundách, ve které lze zprávy ukládat do vyrovnávací paměti před odesláním. Pokud je tento interval dosažen, všechny zprávy ve vyrovnávací paměti se odesílají jako dávka bez ohledu na to, kolik nebo kolik jich je. Pokud není nastavená, výchozí hodnota je 5.	No
`maxMessages`	Maximální počet zpráv, které lze ukládat do vyrovnávací paměti před odesláním. Pokud se toto číslo dosáhne, všechny zprávy ve vyrovnávací paměti se odesílají jako dávka bez ohledu na to, jak velké nebo jak dlouho se ukládají do vyrovnávací paměti. Pokud není nastavená, výchozí hodnota je 1 00000.	No
`maxBytes`	Maximální velikost v bajtech, které lze ukládat do vyrovnávací paměti před odesláním. Pokud je tato velikost dosažená, všechny zprávy ve vyrovnávací paměti se odesílají jako dávka bez ohledu na to, kolik nebo jak dlouho se ukládají do vyrovnávací paměti. Výchozí hodnota je 1 000000 (1 MB).	No

Pokud například nastavíte latenciMs na 1000, maxMessages na 100 a maxBytes na 1024, zprávy se odesílají buď v případě, že je v vyrovnávací paměti 100 zpráv, nebo když je v vyrovnávací paměti 1 024 bajtů nebo když od posledního odeslání uplynulo 1 000 milisekund, podle toho, co nastane dříve.

Konfigurace dávkování:

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole Batching povolte dávkování. Pomocí polí Dávkové latence, Maximální bajty a Počet zpráv určete nastavení dávkování.

kafkaSettings: {
  batching: {
    mode: 'Enabled' // Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

kafkaSettings:
  batching:
    mode: Enabled # Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Toto nastavení se projeví jenom v případě, že se koncový bod použije jako cíl, ve kterém je tok dat producentem.

Strategie zpracování oddílů

Strategie zpracování oddílů řídí, jak se zprávy přiřazují k oddílům Kafka při jejich odesílání do témat Kafka. Oddíly Kafka jsou logické segmenty tématu Kafka, které umožňují paralelní zpracování a odolnost proti chybám. Každá zpráva v tématu Kafka má oddíl a posun, který slouží k identifikaci a řazení zpráv.

Toto nastavení se projeví jenom v případě, že se koncový bod použije jako cíl, ve kterém je tok dat producentem.

Ve výchozím nastavení tok dat přiřazuje zprávy náhodným oddílům pomocí algoritmu kruhového dotazování. Pomocí různých strategií ale můžete přiřadit zprávy k oddílům na základě určitých kritérií, jako je název tématu MQTT nebo vlastnost zprávy MQTT. To vám pomůže dosáhnout lepšího vyrovnávání zatížení, umístění dat nebo řazení zpráv.

Hodnota	Popis
`Default`	Přiřadí zprávy náhodným oddílům pomocí algoritmu kruhového dotazování. Toto je výchozí hodnota, pokud není zadána žádná strategie.
`Static`	Přiřadí zprávy k pevnému číslu oddílu, které je odvozeno z ID instance toku dat. To znamená, že každá instance toku dat odesílá zprávy do jiného oddílu. To může pomoct dosáhnout lepšího vyrovnávání zatížení a umístění dat.
`Topic`	Jako klíč pro dělení použije název tématu MQTT ze zdroje toku dat. To znamená, že zprávy se stejným názvem tématu MQTT se odesílají do stejného oddílu. To může pomoct dosáhnout lepšího pořadí zpráv a umístění dat.
`Property`	Používá vlastnost zprávy MQTT ze zdroje toku dat jako klíč pro dělení. Zadejte název vlastnosti v `partitionKeyProperty` poli. To znamená, že zprávy se stejnou hodnotou vlastnosti se odesílají do stejného oddílu. To může pomoct dosáhnout lepšího pořadí zpráv a umístění dat na základě vlastního kritéria.

Pokud například nastavíte strategii zpracování oddílů na Property klíč oddílu a vlastnost klíče oddílu na device-id, zprávy se stejnou device-id vlastností se odešlou do stejného oddílu.

Konfigurace strategie zpracování oddílů:

Na stránce nastavení koncového bodu toku dat vyberte kartu Upřesnit a pak pomocí pole Strategie zpracování oddílů určete strategii zpracování oddílů. Pomocí pole vlastnosti Klíč oddílu určete vlastnost použitou pro dělení, pokud je strategie nastavena na Property.

kafkaSettings: {
  partitionStrategy: 'Default' // Or Static, Topic, Property
  partitionKeyProperty: '<PROPERTY_NAME>'
}

kafkaSettings:
  partitionStrategy: Default # Or Static, Topic, Property
  partitionKeyProperty: <PROPERTY_NAME>

Potvrzení Kafka

Potvrzení Kafka (acks) slouží k řízení stálosti a konzistence zpráv odesílaných do témat Kafka. Když producent odešle zprávu do tématu Kafka, může požádat o různé úrovně potvrzení od zprostředkovatele Kafka, aby se zajistilo, že se zpráva úspěšně zapíše do tématu a replikuje se v clusteru Kafka.

Toto nastavení se projeví jenom v případě, že se koncový bod použije jako cíl (to znamená, že tok dat je producent).

Hodnota	Popis
`None`	Tok dat nečeká na potvrzení zprostředkovatele Kafka. Toto nastavení je nejrychlejší, ale nejméně odolná možnost.
`All`	Tok dat počká, až se zpráva zapíše do oddílu vodicího znaku a všechny následné oddíly. Toto nastavení je nejpomalejší, ale nejodolnější možností. Toto nastavení je také výchozí možností.
`One`	Tok dat čeká na zápis zprávy do oddílu vodicího znaku a alespoň jeden následný oddíl.
`Zero`	Tok dat čeká, až se zpráva zapíše do oddílu leader, ale nečeká na potvrzení od sledujících. To je rychlejší než `One` méně odolné.

Pokud například nastavíte potvrzení AllKafka, tok dat před odesláním další zprávy počká, až se zpráva zapíše do oddílu vedoucího oddílu a všechny následující oddíly.

Konfiguracepotvrzeních

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole potvrzení Kafka určete úroveň potvrzení Kafka.

kafkaSettings: {
  kafkaAcks: 'All' // Or None, One, Zero
}

kafkaSettings:
  kafkaAcks: All # Or None, One, Zero

Toto nastavení se projeví jenom v případě, že se koncový bod použije jako cíl, ve kterém je tok dat producentem.

Kopírování vlastností MQTT

Ve výchozím nastavení je povolené nastavení vlastností MQTT kopírování. Tyto vlastnosti uživatele zahrnují hodnoty, jako subject je například uložení názvu prostředku odesílajícího zprávu.

Na stránce nastavení koncového bodu toku dat vyberte kartu Upřesnit a potom pomocí zaškrtávacího políčka vedle pole Kopírovat vlastnosti MQTT povolte nebo zakažte kopírování vlastností MQTT.

kafkaSettings: {
  copyMqttProperties: 'Enabled' // Or Disabled
}

kafkaSettings:
  copyMqttProperties: Enabled # Or Disabled

Následující části popisují, jak se vlastnosti MQTT překládají do hlaviček uživatelů Kafka a naopak při povolení nastavení.

Koncový bod Kafka je cíl

Pokud je koncový bod Kafka cílem toku dat, přeloží se všechny definované vlastnosti specifikace MQTT v5 do hlaviček uživatelů Kafka. Například zpráva MQTT v5 s přeposlání typu obsahu do Kafka se přeloží do hlavičky "Content Type":{specifiedValue}uživatele Kafka . Podobná pravidla platí pro další předdefinované vlastnosti MQTT definované v následující tabulce.

MQTT – vlastnost	Přeložené chování
Indikátor formátu datové části	Klíč: Indikátor formátu datové části Hodnota: "0" (datová část je bajty) nebo "1" (datová část je UTF-8)
Téma odpovědi	Klíč: "Téma odpovědi" Hodnota: Kopie tématu odpovědi z původní zprávy.
Interval vypršení platnosti zprávy	Klíč: "Interval vypršení platnosti zprávy" Hodnota: Reprezentace počtu sekund před vypršením platnosti zprávy UTF-8 Další podrobnosti najdete v tématu Vlastnost Interval vypršení platnosti zprávy.
Data korelace:	Klíč: "Korelační data" Hodnota: Kopie dat korelace z původní zprávy. Na rozdíl od mnoha vlastností MQTT v5, které jsou kódovány UTF-8, mohou být korelační data libovolná.
Typ obsahu:	Klíč: "Typ obsahu" Hodnota: Kopie typu obsahu z původní zprávy

Páry hodnot klíče uživatelských vlastností MQTT v5 se přímo překládají do hlaviček uživatelů Kafka. Pokud má záhlaví uživatele ve zprávě stejný název jako předdefinovaná vlastnost MQTT (například hlavička uživatele s názvem Data korelace), pak zda předávání hodnoty vlastnosti specifikace MQTT v5 nebo vlastnost uživatele není definována.

Toky dat nikdy nedostávají tyto vlastnosti od Zprostředkovatele MQTT. Proto je tok dat nikdy nepředá:

Alias tématu
Identifikátory předplatného

Vlastnost Interval vypršení platnosti zprávy

Interval vypršení platnosti zprávy určuje, jak dlouho může zpráva zůstat ve zprostředkovateli MQTT před zahozením.

Když tok dat obdrží zprávu MQTT se zadaným intervalem vypršení platnosti zprávy, bude následující:

Zaznamenává čas přijetí zprávy.
Než se zpráva vygeneruje do cíle, čas se odečte od zprávy do fronty od původního intervalu vypršení platnosti.
Pokud zpráva nevypršela (výše uvedená operace je > 0), zpráva se vygeneruje do cíle a obsahuje aktualizovaný čas vypršení platnosti zprávy.
Pokud vypršela platnost zprávy (výše uvedená operace je <= 0), zpráva se nevygeneruje cílem.

Příklady:

Tok dat obdrží zprávu MQTT s intervalem vypršení platnosti zprávy = 3600 sekund. Odpovídající cíl je dočasně odpojený, ale je schopen se znovu připojit. 1 000 sekund před odesláním této zprávy MQTT do cíle. V tomto případě má zpráva cíle nastavený interval vypršení platnosti zprávy na hodnotu 2600 (3600 – 1000) sekund.
Tok dat obdrží zprávu MQTT s intervalem vypršení platnosti zprávy = 3600 sekund. Odpovídající cíl je dočasně odpojený, ale je schopen se znovu připojit. V takovém případě ale opětovné připojení trvá 4 000 sekund. Platnost zprávy vypršela a tok dat tuto zprávu nepřepošla do cíle.

Koncový bod Kafka je zdroj toku dat.

Poznámka:

Při použití koncového bodu služby Event Hubs jako zdroje toku dat je známý problém, kdy se hlavička Kafka poškodí jako přeloženou do MQTT. K tomu dochází pouze v případě, že používáte centrum událostí přes klienta centra událostí, který používá AMQP v rámci krytu. Například "foo"="bar" se přeloží "foo", ale hodnota se stane"\xa1\x03bar".

Pokud je koncový bod Kafka zdrojem toku dat, přeloží se hlavičky uživatelů Kafka do vlastností MQTT v5. Následující tabulka popisuje, jak se hlavičky uživatelů Kafka překládají do vlastností MQTT v5.

Hlavička Kafka	Přeložené chování
Klíč	Klíč: "Klíč" Hodnota: Kopie klíče z původní zprávy
Časové razítko	Klíč: Časové razítko Hodnota: Kódování UTF-8 časového razítka Kafka, což je počet milisekund od unixového epochy.

Páry klíč/hodnota hlavičky uživatele Kafka – za předpokladu, že jsou všechny kódované v UTF-8 – se přímo překládají do vlastností uživatelského klíče/hodnoty MQTT.

UTF-8 / Binární neshody

MQTT v5 podporuje pouze vlastnosti založené na UTF-8. Pokud tok dat obdrží zprávu Kafka, která obsahuje jednu nebo více hlaviček bez UTF-8, tok dat:

Odeberte vlastnost nebo vlastnosti pro přesměrování.
Přeposlání zbytku zprávy dál podle předchozích pravidel

Aplikace, které vyžadují binární přenos v hlavičce zdroje Kafka => MQTT Target vlastnosti, musí je nejprve zakódovat UTF-8 – například prostřednictvím Base64.

>=Neshoda vlastností 64 kB

Vlastnosti MQTT v5 musí být menší než 64 kB. Pokud tok dat obdrží zprávu Kafka obsahující jednu nebo více hlaviček, která je >= 64 kB, tok dat:

Odeberte vlastnost nebo vlastnosti pro přesměrování.
Přeposlání zbytku zprávy dál podle předchozích pravidel

Překlad vlastností při použití služby Event Hubs a výrobců používajících AMQP

Pokud máte klienta, který předává zprávy zdrojovému koncovému bodu toku dat Kafka, proveďte některou z následujících akcí:

Odesílání zpráv do služby Event Hubs pomocí klientských knihoven, jako je Azure.Messaging.EventHubs
Přímé použití AMQP

Existují drobné odlišnosti překladu vlastností, o které je potřeba vědět.

Měli byste udělat jednu z těchto věcí:

Vyhněte se odesílání vlastností
Pokud musíte odeslat vlastnosti, odešlete hodnoty kódované jako UTF-8.

Když Event Hubs přeloží vlastnosti z AMQP do Kafka, zahrne do zprávy základní typy kódování AMQP. Další informace o chování naleznete v tématu Výměna událostí mezi spotřebiteli a producenty pomocí různých protokolů.

V následujícím příkladu kódu, když koncový bod toku dat obdrží hodnotu "foo":"bar", obdrží vlastnost jako <0xA1 0x03 "bar">.

using global::Azure.Messaging.EventHubs;
using global::Azure.Messaging.EventHubs.Producer;

var propertyEventBody = new BinaryData("payload");

var propertyEventData = new EventData(propertyEventBody)
{
  Properties =
  {
    {"foo", "bar"},
  }
};

var propertyEventAdded = eventBatch.TryAdd(propertyEventData);
await producerClient.SendAsync(eventBatch);

Koncový bod toku dat nemůže předat vlastnost <0xA1 0x03 "bar"> datové části do zprávy MQTT, protože data nejsou UTF-8. Pokud však zadáte řetězec UTF-8, koncový bod toku dat přeloží řetězec před odesláním na MQTT. Pokud použijete řetězec UTF-8, bude zpráva MQTT obsahovat "foo":"bar" vlastnosti uživatele.

Přeloží se jenom hlavičky UTF-8. Například vzhledem k následujícímu scénáři, ve kterém je vlastnost nastavena jako plovoucí:

Properties = 
{
  {"float-value", 11.9 },
}

Koncový bod toku dat zahodí pakety, které obsahují "float-value" pole.

Ne všechny vlastnosti dat události, včetně propertyEventData.correlationId, se přeposílají. Další informace naleznete v tématu Vlastnosti uživatele události,

CloudEvents

CloudEvents představují způsob, jak běžným způsobem popsat data událostí. Nastavení CloudEvents slouží k odesílání nebo přijímání zpráv ve formátu CloudEvents. CloudEvents můžete použít pro architektury řízené událostmi, ve kterých se různé služby potřebují vzájemně komunikovat ve stejných nebo různých poskytovatelích cloudu.

Možnosti CloudEventAttributes jsou Propagate neboCreateOrRemap.

Na stránce nastavení koncového bodu toku dat v provozním prostředí vyberte kartu Upřesnit a pak pomocí pole Atributy cloudové události určete nastavení CloudEvents.

kafkaSettings: {
  cloudEventAttributes: 'Propagate' // Or CreateOrRemap
}

kafkaSettings:
  cloudEventAttributes: Propagate # Or CreateOrRemap

Následující části popisují, jak se vlastnosti CloudEvent šíří nebo vytvářejí a znovu namapují.

Rozšířit nastavení

Vlastnosti CloudEvent se předávají pro zprávy, které obsahují požadované vlastnosti. Pokud zpráva neobsahuje požadované vlastnosti, zpráva se předává tak, jak je. Pokud jsou k dispozici požadované vlastnosti, ce_ přidá se do názvu vlastnosti CloudEvent předpona.

Název	Požaduje se	Ukázková hodnota	Název výstupu	Výstupní hodnota
`specversion`	Ano	`1.0`	`ce-specversion`	Předáno tak, jak je
`type`	Ano	`ms.aio.telemetry`	`ce-type`	Předáno tak, jak je
`source`	Ano	`aio://mycluster/myoven`	`ce-source`	Předáno tak, jak je
`id`	Ano	`A234-1234-1234`	`ce-id`	Předáno tak, jak je
`subject`	No	`aio/myoven/telemetry/temperature`	`ce-subject`	Předáno tak, jak je
`time`	No	`2018-04-05T17:31:00Z`	`ce-time`	Prošlo se tak, jak je. Není převzorkovaná.
`datacontenttype`	No	`application/json`	`ce-datacontenttype`	Změna na typ výstupního datového obsahu po volitelné fázi transformace
`dataschema`	No	`sr://fabrikam-schemas/123123123234234234234234#1.0.0`	`ce-dataschema`	Pokud je v konfiguraci transformace předáno schéma transformace výstupních dat, `dataschema` změní se na výstupní schéma.

Nastavení CreateOrRemap

Vlastnosti CloudEvent se předávají pro zprávy, které obsahují požadované vlastnosti. Pokud zpráva neobsahuje požadované vlastnosti, vygenerují se vlastnosti.

Název	Požaduje se	Název výstupu	Vygenerovaná hodnota, pokud chybí
`specversion`	Ano	`ce-specversion`	`1.0`
`type`	Ano	`ce-type`	`ms.aio-dataflow.telemetry`
`source`	Ano	`ce-source`	`aio://<target-name>`
`id`	Yes	`ce-id`	Vygenerované UUID v cílovém klientovi
`subject`	No	`ce-subject`	Výstupní téma, ve kterém je zpráva odeslána
`time`	No	`ce-time`	Vygenerováno jako RFC 3339 v cílovém klientovi
`datacontenttype`	No	`ce-datacontenttype`	Změna na typ výstupního datového obsahu po volitelné fázi transformace
`dataschema`	No	`ce-dataschema`	Schéma definované v registru schématu

Další kroky

Další informace o tocích dat najdete v tématu Vytvoření toku dat.

Sdílet prostřednictvím

Konfigurace koncových bodů toku dat Azure Event Hubs a Kafka

Požadavky

Azure Event Hubs

Vytvoření oboru názvů a centra událostí Služby Azure Event Hubs

Přiřazení oprávnění ke spravované identitě

Vytvoření koncového bodu toku dat pro Azure Event Hubs

Použití připojovací řetězec pro ověřování ve službě Event Hubs

Omezení

Vlastní zprostředkovatelé Kafka

Dostupné metody ověřování

Spravovaná identita přiřazená systémem

Spravovaná identita přiřazená uživatelem

SASL

Anonymní

Rozšířené nastavení

Nastavení protokolu TLS

Režim TLS

Certifikát důvěryhodné certifikační autority

ID skupiny příjemců

Komprese

Dávkování

Strategie zpracování oddílů

Potvrzení Kafka

Kopírování vlastností MQTT

Koncový bod Kafka je cíl

Vlastnost Interval vypršení platnosti zprávy

Koncový bod Kafka je zdroj toku dat.

UTF-8 / Binární neshody

>=Neshoda vlastností 64 kB

Překlad vlastností při použití služby Event Hubs a výrobců používajících AMQP

CloudEvents

Rozšířit nastavení

Nastavení CreateOrRemap

Další kroky

Váš názor

Další materiály