Azure Sanal Ağı üzerinden HDInsight üzerinde Apache Kafka'ya bağlanma

Makale
06/14/2024

Azure Sanal Ağ aracılığıyla HDInsight üzerinde Apache Kafka'ya doğrudan bağlanmayı öğrenin. Bu belge, aşağıdaki yapılandırmaları kullanarak Kafka'ya bağlanma hakkında bilgi sağlar:

Şirket içi ağdaki kaynaklardan. Bu bağlantı, yerel ağınızda bir VPN cihazı (yazılım veya donanım) kullanılarak kurulur.
VPN yazılım istemcisi kullanan bir geliştirme ortamından.

Not

Azure ile etkileşim kurmak için Azure Az PowerShell modülünü kullanmanızı öneririz. Başlamak için bkz . Azure PowerShell'i yükleme. Az PowerShell modülüne nasıl geçeceğinizi öğrenmek için bkz. Azure PowerShell’i AzureRM’den Az’ye geçirme.

Mimari ve planlama

HDInsight, genel İnternet üzerinden Kafka'ya doğrudan bağlantıya izin vermez. Bunun yerine, Kafka istemcileri (üreticiler ve tüketiciler) aşağıdaki bağlantı yöntemlerinden birini kullanmalıdır:

İstemciyi HDInsight üzerinde Kafka ile aynı sanal ağda çalıştırın. Bu yapılandırma, HDInsight üzerinde Apache Kafka ile başlat belgesinde kullanılır. İstemci doğrudan HDInsight küme düğümlerinde veya aynı ağdaki başka bir sanal makinede çalışır.
Şirket içi ağınız gibi özel bir ağı sanal ağa Bağlan. Bu yapılandırma, şirket içi ağınızdaki istemcilerin kafka ile doğrudan çalışmasını sağlar. Bu yapılandırmayı etkinleştirmek için aşağıdaki görevleri gerçekleştirin:
1. Özel ağ oluşturun.
2. Siteden siteye yapılandırma kullanan bir VPN ağ geçidi oluşturun. Bu belgede kullanılan yapılandırma, şirket içi ağınızdaki bir VPN ağ geçidi cihazına bağlanır.
3. Sanal ağda bir DNS sunucusu oluşturun.
4. Her ağdaki DNS sunucusu arasında iletmeyi yapılandırın.
5. Sanal ağda HDInsight kümesinde Kafka oluşturun.
  
  Daha fazla bilgi için, şirket içi ağdan Apache Kafka'ya Bağlan bölümüne bakın.
VPN ağ geçidi ve VPN istemcisi kullanarak tek tek makineleri sanal ağa Bağlan. Bu yapılandırmayı etkinleştirmek için aşağıdaki görevleri gerçekleştirin:
1. Özel ağ oluşturun.
2. Noktadan siteye yapılandırma kullanan bir VPN ağ geçidi oluşturun. Bu yapılandırma hem Windows hem de MacOS istemcileriyle kullanılabilir.
3. Sanal ağda HDInsight kümesinde Kafka oluşturun.
4. KAFKA'yi IP reklamları için yapılandırın. Bu yapılandırma, istemcinin etki alanı adları yerine aracı IP adreslerini kullanarak bağlanmasına olanak tanır.
5. Geliştirme sisteminde VPN istemcisini indirin ve kullanın.
  
  Daha fazla bilgi için VPN istemcisi ile Apache Kafka'ya Bağlan bölümüne bakın.
  Uyarı
  
  Bu yapılandırma, aşağıdaki sınırlamalar nedeniyle yalnızca geliştirme amacıyla önerilir:
  - Her istemcinin bir VPN yazılım istemcisi kullanarak bağlanması gerekir.
  - VPN istemcisi, ad çözümleme isteklerini sanal ağa geçirmediğinden, Kafka ile iletişim kurmak için IP adreslemesi kullanmanız gerekir. IP iletişimi Kafka kümesinde ek yapılandırma gerektirir.

Sanal ağda HDInsight kullanma hakkında daha fazla bilgi için bkz . Azure HDInsight kümeleri için sanal ağ planlama.

Şirket içi ağdan Apache Kafka'ya Bağlan

Şirket içi ağınızla iletişim kuran bir Kafka kümesi oluşturmak için HDInsight'ı şirket içi ağ belgenize Bağlan adımlarını izleyin.

Önemli

HDInsight kümesini oluştururken Kafka küme türünü seçin.

Bu adımlar aşağıdaki yapılandırmayı oluşturur:

Azure Sanal Ağ
Siteden siteye VPN ağ geçidi
Azure Depolama hesabı (HDInsight tarafından kullanılır)
HDInsight üzerinde Kafka

Kafka istemcisinin şirket içinden kümeye bağlanabildiğini doğrulamak için Örnek: Python istemcisi bölümündeki adımları kullanın.

VPN istemcisiyle Apache Kafka'ya Bağlan

Aşağıdaki yapılandırmayı oluşturmak için bu bölümdeki adımları kullanın:

Azure Sanal Ağ
Noktadan siteye VPN ağ geçidi
Azure Depolama Hesabı (HDInsight tarafından kullanılır)
HDInsight üzerinde Kafka

Noktadan siteye bağlantılar için otomatik olarak imzalanan sertifikalarla çalışma belgesindeki adımları izleyin. Bu belge, ağ geçidi için gereken sertifikaları oluşturur.

Bir PowerShell istemi açın ve Azure aboneliğinizde oturum açmak için aşağıdaki kodu kullanın:

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

Yapılandırma bilgilerini içeren değişkenler oluşturmak için aşağıdaki kodu kullanın:

# 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"

Azure kaynak grubunu ve sanal ağı oluşturmak için aşağıdaki kodu kullanın:

# 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

Uyarı

Bu işlemin tamamlanması birkaç dakika sürebilir.

Azure Depolama Hesabı ve blob kapsayıcısını oluşturmak için aşağıdaki kodu kullanın:

# 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

HDInsight kümesini oluşturmak için aşağıdaki kodu kullanın:

# 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

Uyarı

Bu işlemin tamamlanması yaklaşık 15 dakika sürer.

KAFKA'yi IP reklamları için yapılandırma

Varsayılan olarak Apache Zookeeper, Kafka aracılarının etki alanı adını istemcilere döndürür. Sanal ağdaki varlıklar için ad çözümlemesi kullanılamadığından bu yapılandırma VPN yazılım istemcisiyle çalışmaz. Bu yapılandırmada, Kafka'yı etki alanı adları yerine IP adreslerini tanıtacak şekilde yapılandırmak için aşağıdaki adımları kullanın:

Web tarayıcısını kullanarak https://CLUSTERNAME.azurehdinsight.net adresine gidin. değerini HDInsight kümesindeki Kafka adıyla değiştirin CLUSTERNAME .

İstendiğinde, küme için HTTPS kullanıcı adını ve parolasını kullanın. Küme için Ambari Web kullanıcı arabirimi görüntülenir.
Kafka'da bilgileri görüntülemek için soldaki listeden Kafka'yı seçin.
Kafka yapılandırmasını görüntülemek için üst ortadaki Yapılandırmalar'ı seçin.
kafka-env yapılandırmasını bulmak için sağ üstteki Filtre alanına girinkafka-env.

Kafka'yı IP adreslerini tanıtacak şekilde yapılandırmak için kafka-env-template alanının altına aşağıdaki metni ekleyin:

# 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

Kafka'nın dinlediğini arabirimi yapılandırmak için sağ üstteki Filtre alanına girinlisteners.
Kafka'yı tüm ağ arabirimlerinde dinleyecek şekilde yapılandırmak için dinleyiciler alanındaki değeri olarak PLAINTEXT://0.0.0.0:9092değiştirin.
Yapılandırma değişikliklerini kaydetmek için Kaydet düğmesini kullanın. Değişiklikleri açıklayan bir kısa mesaj girin. Değişiklikler kaydedildikten sonra Tamam'ı seçin.
Kafka'yı yeniden başlatırken hataları önlemek için Hizmet Eylemleri düğmesini kullanın ve Bakım Modunu Aç'ı seçin. Bu işlemi tamamlamak için Tamam'ı seçin.
Kafka'yı yeniden başlatmak için Yeniden Başlat düğmesini kullanın ve Tüm Etkilenenleri Yeniden Başlat'ı seçin. Yeniden başlatmayı onaylayın ve işlem tamamlandıktan sonra Tamam düğmesini kullanın.
Bakım modunu devre dışı bırakmak için Hizmet Eylemleri düğmesini kullanın ve Bakım Modunu Kapat'ı seçin. Bu işlemi tamamlamak için Tamam'ı seçin.

VPN ağ geçidine Bağlan

VPN ağ geçidine bağlanmak için Noktadan Siteye bağlantı yapılandırma belgesinin azurea Bağlan bölümünü kullanın.

Örnek: Python istemcisi

Kafka bağlantısını doğrulamak için python üreticisi ve tüketicisi oluşturmak ve çalıştırmak için aşağıdaki adımları kullanın:

Kafka kümesindeki düğümlerin tam etki alanı adını (FQDN) ve IP adreslerini almak için aşağıdaki yöntemlerden birini kullanın:

$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}"

Bu betik, sanal ağı içeren Azure kaynak grubunun adı olduğunu varsayar $resourceGroupName .

Döndürülen bilgileri sonraki adımlarda kullanmak üzere kaydedin.

Kafka-python istemcisini yüklemek için aşağıdakileri kullanın:
```
pip install kafka-python
```
Kafka'ya veri göndermek için aşağıdaki Python kodunu kullanın:
```
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' Girdileri, bu bölümdeki 1. adımdan döndürülen adreslerle değiştirin:
- Yazılım VPN istemcisi kullanıyorsanız, girdileri çalışan düğümlerinizin IP adresiyle değiştirinkafka_broker.
- Özel bir DNS sunucusu aracılığıyla ad çözümlemeyi kafka_broker etkinleştirdiyseniz, girdileri çalışan düğümlerinin FQDN'siyle değiştirin.
  
  Not
  
  Bu kod dizesini test message konusuna testtopicgönderir. HDInsight'ta Kafka'nın varsayılan yapılandırması, konu yoksa oluşturulmamasıdır. Bkz. Konuları otomatik olarak oluşturmak için HDInsight üzerinde Apache Kafka'yı yapılandırma. Alternatif olarak, iletileri oluşturmadan önce konuları el ile oluşturabilirsiniz.

Kafka'dan iletileri almak için aşağıdaki Python kodunu kullanın:

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' Girdileri, bu bölümdeki 1. adımdan döndürülen adreslerle değiştirin:

Yazılım VPN istemcisi kullanıyorsanız, girdileri çalışan düğümlerinizin IP adresiyle değiştirinkafka_broker.
Özel bir DNS sunucusu aracılığıyla ad çözümlemeyi kafka_broker etkinleştirdiyseniz, girdileri çalışan düğümlerinin FQDN'siyle değiştirin.

Sonraki adımlar

HDInsight'ı sanal ağ ile kullanma hakkında daha fazla bilgi için Azure HDInsight kümeleri için sanal ağ dağıtımı planlama belgesine bakın.

Noktadan Siteye VPN ağ geçidi ile Azure Sanal Ağ oluşturma hakkında daha fazla bilgi için aşağıdaki belgelere bakın:

HDInsight üzerinde Apache Kafka ile çalışma hakkında daha fazla bilgi için aşağıdaki belgelere bakın:

Aracılığıyla paylaş