Připojení k Apache Kafka ve službě HDInsight přes virtuální síť Azure

Článek
03/19/2024

Zjistěte, jak se přímo připojit k Apache Kafka ve službě HDInsight prostřednictvím virtuální sítě Azure. Tento dokument obsahuje informace o připojení k Systému Kafka pomocí následujících konfigurací:

Z prostředků v místní síti. Toto připojení se naváže pomocí zařízení VPN (softwaru nebo hardwaru) ve vaší místní síti.
Z vývojového prostředí pomocí softwarového klienta VPN.

Poznámka:

Při práci s Azure doporučujeme používat modul Azure Az PowerShellu. Začněte tím, že si projdete téma Instalace Azure PowerShellu. Informace o tom, jak migrovat na modul Az PowerShell, najdete v tématu Migrace Azure PowerShellu z AzureRM na Az.

Architektura a plánování

HDInsight neumožňuje přímé připojení k Systému Kafka přes veřejný internet. Místo toho musí klienti Kafka (producenti a příjemci) používat jednu z následujících metod připojení:

Spusťte klienta ve stejné virtuální síti jako Kafka ve službě HDInsight. Tato konfigurace se používá v dokumentu Start s Apache Kafka ve službě HDInsight . Klient běží přímo na uzlech clusteru HDInsight nebo na jiném virtuálním počítači ve stejné síti.
Připojení privátní síti, jako je vaše místní síť, do virtuální sítě. Tato konfigurace umožňuje klientům ve vaší místní síti přímo pracovat se systémem Kafka. Pokud chcete tuto konfiguraci povolit, proveďte následující úlohy:
1. Vytvořte virtuální síť.
2. Vytvořte bránu VPN, která používá konfiguraci typu site-to-site. Konfigurace použitá v tomto dokumentu se připojuje k zařízení brány VPN ve vaší místní síti.
3. Vytvořte ve virtuální síti server DNS.
4. Nakonfigurujte předávání mezi serverem DNS v každé síti.
5. Ve virtuální síti vytvořte cluster Kafka ve službě HDInsight.
  
  Další informace najdete v části Připojení k Apache Kafka z části místní sítě.
Připojení jednotlivé počítače do virtuální sítě pomocí brány VPN a klienta VPN. Pokud chcete tuto konfiguraci povolit, proveďte následující úlohy:
1. Vytvořte virtuální síť.
2. Vytvořte bránu VPN, která používá konfiguraci typu point-to-site. Tuto konfiguraci je možné použít s klienty Windows i MacOS.
3. Ve virtuální síti vytvořte cluster Kafka ve službě HDInsight.
4. Nakonfigurujte Kafka pro inzerci IP. Tato konfigurace umožňuje klientovi připojit se místo názvů domén pomocí IP adres zprostředkovatele.
5. Stáhněte a použijte klienta VPN ve vývojovém systému.
  
  Další informace najdete v oddílu Připojení apache Kafka s klientem VPN.
  Upozorňující
  
  Tato konfigurace se doporučuje pouze pro účely vývoje z důvodu následujících omezení:
  - Každý klient se musí připojit pomocí softwarového klienta VPN.
  - Klient VPN nepředává do virtuální sítě požadavky na překlad ip adres, takže pro komunikaci se systémem Kafka musíte použít přidělování IP adres. Komunikace IP vyžaduje další konfiguraci v clusteru Kafka.

Další informace o používání služby HDInsight ve virtuální síti najdete v tématu Plánování virtuální sítě pro clustery Azure HDInsight.

Připojení k Apache Kafka z místní sítě

Pokud chcete vytvořit cluster Kafka, který komunikuje s vaší místní sítí, postupujte podle kroků v Připojení HDInsight k místnímu síťovému dokumentu.

Důležité

Při vytváření clusteru HDInsight vyberte typ clusteru Kafka .

Tento postup vytvoří následující konfiguraci:

Azure Virtual Network
Brána VPN typu Site-to-Site
Účet služby Azure Storage (používaný službou HDInsight)
Kafka v HDInsightu

Pokud chcete ověřit, že se klient Kafka může připojit ke clusteru z místního prostředí, postupujte podle kroků v části Příklad: Klient Python.

Připojení k Apache Kafka pomocí klienta VPN

Pomocí kroků v této části vytvořte následující konfiguraci:

Azure Virtual Network
Brána VPN typu Point-to-Site
Účet služby Azure Storage (používaný službou HDInsight)
Kafka v HDInsightu

Postupujte podle kroků v dokumentu Práce s certifikáty podepsanými svým držitelem pro dokument připojení typu Point-to-Site. Tento dokument vytvoří certifikáty potřebné pro bránu.

Otevřete příkazový řádek PowerShellu a pomocí následujícího kódu se přihlaste ke svému předplatnému Azure:

Connect-AzAccount
# If you have multiple subscriptions, uncomment to set the subscription
#Select-AzSubscription -SubscriptionName "name of your subscription"

Pomocí následujícího kódu vytvořte proměnné, které obsahují informace o konfiguraci:

# Prompt for generic information
$resourceGroupName = Read-Host "What is the resource group name?"
$baseName = Read-Host "What is the base name? It is used to create names for resources, such as 'net-basename' and 'kafka-basename':"
$location = Read-Host "What Azure Region do you want to create the resources in?"
$rootCert = Read-Host "What is the file path to the root certificate? It is used to secure the VPN gateway."

# Prompt for HDInsight credentials
$adminCreds = Get-Credential -Message "Enter the HTTPS user name and password for the HDInsight cluster" -UserName "admin"
$sshCreds = Get-Credential -Message "Enter the SSH user name and password for the HDInsight cluster" -UserName "sshuser"

# Names for Azure resources
$networkName = "net-$baseName"
$clusterName = "kafka-$baseName"
$storageName = "store$baseName" # Can't use dashes in storage names
$defaultContainerName = $clusterName
$defaultSubnetName = "default"
$gatewaySubnetName = "GatewaySubnet"
$gatewayPublicIpName = "GatewayIp"
$gatewayIpConfigName = "GatewayConfig"
$vpnRootCertName = "rootcert"
$vpnName = "VPNGateway"

# Network settings
$networkAddressPrefix = "10.0.0.0/16"
$defaultSubnetPrefix = "10.0.0.0/24"
$gatewaySubnetPrefix = "10.0.1.0/24"
$vpnClientAddressPool = "172.16.201.0/24"

# HDInsight settings
$hdiWorkerNodes = 4
$hdiVersion = "3.6"
$hdiType = "Kafka"

Pomocí následujícího kódu vytvořte skupinu prostředků Azure a virtuální síť:

# Create the resource group that contains everything
New-AzResourceGroup -Name $resourceGroupName -Location $location

# Create the subnet configuration
$defaultSubnetConfig = New-AzVirtualNetworkSubnetConfig -Name $defaultSubnetName `
    -AddressPrefix $defaultSubnetPrefix
$gatewaySubnetConfig = New-AzVirtualNetworkSubnetConfig -Name $gatewaySubnetName `
    -AddressPrefix $gatewaySubnetPrefix

# Create the subnet
New-AzVirtualNetwork -Name $networkName `
    -ResourceGroupName $resourceGroupName `
    -Location $location `
    -AddressPrefix $networkAddressPrefix `
    -Subnet $defaultSubnetConfig, $gatewaySubnetConfig

# Get the network & subnet that were created
$network = Get-AzVirtualNetwork -Name $networkName `
    -ResourceGroupName $resourceGroupName
$gatewaySubnet = Get-AzVirtualNetworkSubnetConfig -Name $gatewaySubnetName `
    -VirtualNetwork $network
$defaultSubnet = Get-AzVirtualNetworkSubnetConfig -Name $defaultSubnetName `
    -VirtualNetwork $network

# Set a dynamic public IP address for the gateway subnet
$gatewayPublicIp = New-AzPublicIpAddress -Name $gatewayPublicIpName `
    -ResourceGroupName $resourceGroupName `
    -Location $location `
    -AllocationMethod Dynamic
$gatewayIpConfig = New-AzVirtualNetworkGatewayIpConfig -Name $gatewayIpConfigName `
    -Subnet $gatewaySubnet `
    -PublicIpAddress $gatewayPublicIp

# Get the certificate info
# Get the full path in case a relative path was passed
$rootCertFile = Get-ChildItem $rootCert
$cert = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2($rootCertFile)
$certBase64 = [System.Convert]::ToBase64String($cert.RawData)
$p2sRootCert = New-AzVpnClientRootCertificate -Name $vpnRootCertName `
    -PublicCertData $certBase64

# Create the VPN gateway
New-AzVirtualNetworkGateway -Name $vpnName `
    -ResourceGroupName $resourceGroupName `
    -Location $location `
    -IpConfigurations $gatewayIpConfig `
    -GatewayType Vpn `
    -VpnType RouteBased `
    -EnableBgp $false `
    -GatewaySku Standard `
    -VpnClientAddressPool $vpnClientAddressPool `
    -VpnClientRootCertificates $p2sRootCert

Upozorňující

Dokončení tohoto procesu může trvat několik minut.

Pomocí následujícího kódu vytvořte účet úložiště Azure a kontejner objektů blob:

# Create the storage account
New-AzStorageAccount `
    -ResourceGroupName $resourceGroupName `
    -Name $storageName `
    -SkuName Standard_GRS `
    -Location $location `
    -Kind StorageV2 `
    -EnableHttpsTrafficOnly 1

# Get the storage account keys and create a context
$defaultStorageKey = (Get-AzStorageAccountKey -ResourceGroupName $resourceGroupName `
    -Name $storageName)[0].Value
$storageContext = New-AzStorageContext -StorageAccountName $storageName `
    -StorageAccountKey $defaultStorageKey

# Create the default storage container
New-AzStorageContainer -Name $defaultContainerName `
    -Context $storageContext

Pomocí následujícího kódu vytvořte cluster HDInsight:

# Create the HDInsight cluster
New-AzHDInsightCluster `
    -ResourceGroupName $resourceGroupName `
    -ClusterName $clusterName `
    -Location $location `
    -ClusterSizeInNodes $hdiWorkerNodes `
    -ClusterType $hdiType `
    -OSType Linux `
    -Version $hdiVersion `
    -HttpCredential $adminCreds `
    -SshCredential $sshCreds `
    -DefaultStorageAccountName "$storageName.blob.core.windows.net" `
    -DefaultStorageAccountKey $defaultStorageKey `
    -DefaultStorageContainer $defaultContainerName `
    -DisksPerWorkerNode 2 `
    -VirtualNetworkId $network.Id `
    -SubnetName $defaultSubnet.Id

Upozorňující

Dokončení tohoto procesu trvá přibližně 15 minut.

Konfigurace Kafka pro inzerci IP adres

Ve výchozím nastavení vrátí Apache Zookeeper klientům název domény zprostředkovatelů Kafka. Tato konfigurace nefunguje se softwarovým klientem VPN, protože nemůže používat překlad názvů pro entity ve virtuální síti. Pro tuto konfiguraci pomocí následujícího postupu nakonfigurujte Kafka tak, aby místo názvů domén inzerovali IP adresy:

Pomocí webového prohlížeče přejděte na https://CLUSTERNAME.azurehdinsight.net. Nahraďte CLUSTERNAME názvem Kafka v clusteru HDInsight.

Po zobrazení výzvy použijte uživatelské jméno a heslo HTTPS pro cluster. Zobrazí se webové uživatelské rozhraní Ambari pro cluster.
Pokud chcete zobrazit informace o systému Kafka, vyberte v seznamu na levé straně systém Kafka .
Konfiguraci Kafka zobrazíte tak, že v horní části vyberete Konfigurace .
Pokud chcete najít konfiguraci kafka-env , zadejte kafka-env do pole Filtr v pravém horním rohu.

Pokud chcete nakonfigurovat Kafka pro inzerování IP adres, přidejte do dolní části pole šablony kafka-env následující text:

# Configure Kafka to advertise IP addresses instead of FQDN
IP_ADDRESS=$(hostname -i)
echo advertised.listeners=$IP_ADDRESS
sed -i.bak -e '/advertised/{/advertised@/!d;}' /usr/hdp/current/kafka-broker/conf/server.properties
echo "advertised.listeners=PLAINTEXT://$IP_ADDRESS:9092" >> /usr/hdp/current/kafka-broker/conf/server.properties

Pokud chcete nakonfigurovat rozhraní, které Kafka naslouchá, zadejte listenersdo pole Filtr v pravém horním rohu.
Pokud chcete nakonfigurovat Kafka tak, aby naslouchal na všech síťových rozhraních, změňte hodnotu v poli naslouchacího procesu na PLAINTEXT://0.0.0.0:9092hodnotu .
Pokud chcete uložit změny konfigurace, použijte tlačítko Uložit . Zadejte textovou zprávu popisující změny. Po uložení změn vyberte OK .
Pokud chcete zabránit chybám při restartování Systému Kafka, použijte tlačítko Akce služby a vyberte Zapnout režim údržby. Chcete-li dokončit tuto operaci, vyberte OK.
Pokud chcete restartovat Systém Kafka, použijte tlačítko Restartovat a vyberte Restartovat vše, co se týká. Potvrďte restartování a po dokončení operace použijte tlačítko OK .
Pokud chcete režim údržby zakázat, použijte tlačítko Akce služby a vyberte Vypnout režim údržby. Chcete-li dokončit tuto operaci, vyberte OK .

Připojení k bráně VPN

Pokud se chcete připojit k bráně VPN, použijte část Připojení do Azure v dokumentu Konfigurace připojení typu Point-to-Site.

Příklad: Klient Pythonu

Pokud chcete ověřit připojení k Systému Kafka, vytvořte a spusťte producenta a příjemce Pythonu pomocí následujícího postupu:

Pomocí jedné z následujících metod načtěte plně kvalifikovaný název domény (FQDN) a IP adresy uzlů v clusteru Kafka:

$resourceGroupName = "The resource group that contains the virtual network used with HDInsight"

$clusterNICs = Get-AzNetworkInterface -ResourceGroupName $resourceGroupName | where-object {$_.Name -like "*node*"}

$nodes = @()
foreach($nic in $clusterNICs) {
    $node = new-object System.Object
    $node | add-member -MemberType NoteProperty -name "Type" -value $nic.Name.Split('-')[1]
    $node | add-member -MemberType NoteProperty -name "InternalIP" -value $nic.IpConfigurations.PrivateIpAddress
    $node | add-member -MemberType NoteProperty -name "InternalFQDN" -value $nic.DnsSettings.InternalFqdn
    $nodes += $node
}
$nodes | sort-object Type

az network nic list --resource-group <resourcegroupname> --output table --query "[?contains(name,'node')].{NICname:name,InternalIP:ipConfigurations[0].privateIpAddress,InternalFQDN:dnsSettings.internalFqdn}"

Tento skript předpokládá, že $resourceGroupName se jedná o název skupiny prostředků Azure, která obsahuje virtuální síť.

Uložte vrácené informace pro použití v dalších krocích.

Pomocí následujícího příkazu nainstalujte klienta kafka-python :
```
pip install kafka-python
```
Pokud chcete odesílat data do Kafka, použijte následující kód Pythonu:
```
from kafka import KafkaProducer
# Replace the `ip_address` entries with the IP address of your worker nodes
# NOTE: you don't need the full list of worker nodes, just one or two.
producer = KafkaProducer(bootstrap_servers=['kafka_broker_1','kafka_broker_2'])
for _ in range(50):
   producer.send('testtopic', b'test message')
```
'kafka_broker' Položky nahraďte adresami vrácenými z kroku 1 v této části:
- Pokud používáte softwarového klienta VPN, nahraďte kafka_broker položky IP adresou pracovních uzlů.
- Pokud jste povolili překlad názvů prostřednictvím vlastního serveru DNS, nahraďte kafka_broker položky plně kvalifikovaným názvem domény pracovních uzlů.
  
  Poznámka:
  
  Tento kód odešle řetězec test message do tématu testtopic. Výchozí konfigurace Kafka ve službě HDInsight není vytvoření tématu, pokud neexistuje. Viz Postup konfigurace Apache Kafka ve službě HDInsight pro automatické vytváření témat. Další možností je vytvořit témata ručně před vytvářením zpráv.

Pokud chcete načíst zprávy ze systému Kafka, použijte následující kód Pythonu:

from kafka import KafkaConsumer
# Replace the `ip_address` entries with the IP address of your worker nodes
# Again, you only need one or two, not the full list.
# Note: auto_offset_reset='earliest' resets the starting offset to the beginning
#       of the topic
consumer = KafkaConsumer(bootstrap_servers=['kafka_broker_1','kafka_broker_2'],auto_offset_reset='earliest')
consumer.subscribe(['testtopic'])
for msg in consumer:
  print (msg)

'kafka_broker' Položky nahraďte adresami vrácenými z kroku 1 v této části:

Pokud používáte softwarového klienta VPN, nahraďte kafka_broker položky IP adresou pracovních uzlů.
Pokud jste povolili překlad názvů prostřednictvím vlastního serveru DNS, nahraďte kafka_broker položky plně kvalifikovaným názvem domény pracovních uzlů.

Další kroky

Další informace o používání služby HDInsight s virtuální sítí najdete v dokumentu Plánování nasazení virtuální sítě pro clustery Azure HDInsight.

Další informace o vytvoření virtuální sítě Azure s bránou VPN typu Point-to-Site najdete v následujících dokumentech:

Další informace o práci s Apache Kafka ve službě HDInsight najdete v následujících dokumentech: