Взаимодействие с кластерами Apache Kafka в Azure HDInsight через прокси-сервер REST
Прокси-сервер REST для Kafka позволяет взаимодействовать с кластером Kafka через REST API по протоколу HTTPS. Это действие означает, что клиенты Kafka могут находиться за пределами виртуальной сети. Клиенты смогут выполнять простые и защищенные вызовы HTTPS к кластеру Kafka, не используя библиотеки Kafka. В этой статье показано, как создать кластер Kafka с поддержкой прокси-сервера REST. Также вы получите пример кода для вызовов к прокси-серверу REST.
Справочник по REST API
Сведения об операциях, поддерживаемых REST API Kafka, см. в справочнике по API прокси-сервера REST для Kafka в Azure HDInsight.
Общие сведения
Полную спецификацию операций, поддерживаемых API прокси-сервера REST для Apache Kafka, см. здесь.
Конечная точка прокси-сервера REST
При создании кластера Kafka в HDInsight с прокси-сервером REST для этого кластера создается новая общедоступная конечная точка, которую можно найти на странице Свойства для кластера HDInsight на портале Azure.
Безопасность
Доступ к прокси-серверу REST Kafka, управляемому с помощью групп безопасности Microsoft Entra. При создании кластера Kafka предоставьте группе безопасности Microsoft Entra доступ к конечной точке REST. Клиенты Kafka, которым нужен доступ к прокси-серверу REST, должны быть зарегистрированы в этой группе с ролью владельца. Владелец группы может зарегистрироваться через портал или с помощью PowerShell.
Для запросов к конечной точке прокси-сервера REST клиентским приложениям нужно получить токен OAuth. Маркер используется для проверки членства в группах безопасности. В примере клиентского приложения показано, как получить маркер OAuth. Клиентское приложение передает этот токен OAuth прокси-серверу REST в запросе HTTPS.
Примечание.
Дополнительные сведения о группах безопасности Microsoft Entra см. в статье "Управление доступом к приложениям и ресурсам с помощью групп microsoft Entra". Дополнительные сведения о том, как работают маркеры OAuth, см. в статье "Авторизация доступа к веб-приложениям Microsoft Entra" с помощью потока предоставления кода OAuth 2.0.
Работа прокси-сервера REST для Kafka с группами безопасности сети
Если вы используете собственную виртуальную сеть и управляете сетевым трафиком с помощью групп безопасности сети, разрешите входящий трафик через порт 9400 в дополнение к стандартному порту 443. Это гарантирует, что прокси-сервер Kafka REST доступен.
Необходимые компоненты
Регистрация приложения с Microsoft Entra ID. Клиентские приложения, которые записываются для взаимодействия с прокси-сервером REST Kafka, используют идентификатор и секрет этого приложения для проверки подлинности в Azure.
Создайте группу безопасности Microsoft Entra. Добавьте приложение, зарегистрированное с идентификатором Microsoft Entra, в группу безопасности в качестве члена группы. Эта группа безопасности будет использоваться для управления тем, какие приложения позволяют взаимодействовать с прокси-сервером REST. Дополнительные сведения о создании групп Microsoft Entra см. в статье "Создание базовой группы" и добавление участников с помощью идентификатора Microsoft Entra.
Убедитесь, что для этой группы установлен тип Безопасность.
Убедитесь, что приложение является членом группы.
Создание кластера Kafka с включенным прокси-сервером REST
Шаги используют портал Azure. Пример работы с Azure CLI см. в статье Создание кластера Apache Kafka с прокси-сервером REST с помощью Azure CLI.
В процессе создания кластера Kafka установите на вкладке Безопасность и сеть флажок Включить прокси-сервер REST для Kafka.
Щелкните Выбор группы безопасности. Из предложенного списка выберите ту группу безопасности, которая должна иметь доступ к прокси-серверу REST. Чтобы найти нужную группу безопасности, можно воспользоваться полем поиска. В нижней части страницы нажмите кнопку Выбрать.
Выполните остальные действия для создания кластера, как описано в статье Создание кластера Apache Kafka в Azure HDInsight с помощью портала Azure.
После создания кластера перейдите к свойствам кластера, чтобы записать URL-адрес прокси-сервера REST для Kafka.
Пример клиентского приложения
Код Python можно использовать для взаимодействия с прокси-сервером REST в кластере Kafka. Чтобы использовать этот пример кода, выполните следующие шаги:
Сохраните пример кода на компьютере, где установлен Python.
Установите необходимые зависимости Python, выполнив команду
pip3 install msal.Измените раздел Configure these properties (Настройте эти свойства) в коде, указав для следующих свойства значения из вашей среды:
Свойство Description Идентификатор клиента Клиент Azure, на котором находится ваша подписка. Client ID Идентификатор приложения, зарегистрированного в группе безопасности. Секрет клиента Секрет приложения, зарегистрированного в группе безопасности. Kafkarest_endpoint Получите это значение на вкладке Свойства в разделе сводной информации о кластере, как описано в разделе о развертывании. Оно должно быть указано в формате https://<clustername>-kafkarest.azurehdinsight.net.В командной строке выполните файл Python с помощью команды
sudo python3 <filename.py>.
Этот код делает следующее:
- Извлекает маркер OAuth из идентификатора Microsoft Entra.
- Демонстрирует выполнение запроса к прокси-серверу REST для Kafka.
Дополнительные сведения о получении токенов OAuth в Python см. в описании класса AuthenticationContext в Python. Может появиться задержка, пока topics она не создана или удалена через прокси-сервер REST Kafka. Она обусловлена обновлением кэша. Расширено поле value API производителя. Теперь он принимает объекты JSON и любые сериализованные формы.
#Required Python packages
#pip3 install msal
import json
import msal
import random
import requests
import string
import sys
import time
def get_random_string():
letters = string.ascii_letters
random_string = ''.join(random.choice(letters) for i in range(7))
return random_string
#--------------------------Configure these properties-------------------------------#
# Tenant ID for your Azure Subscription
tenant_id = 'ABCDEFGH-1234-1234-1234-ABCDEFGHIJKL'
# Your Client Application Id
client_id = 'XYZABCDE-1234-1234-1234-ABCDEFGHIJKL'
# Your Client Credentials
client_secret = 'password'
# kafka rest proxy -endpoint
kafkarest_endpoint = "https://<clustername>-kafkarest.azurehdinsight.net"
#--------------------------Configure these properties-------------------------------#
# Get access token
# Scope
scope = 'https://hib.azurehdinsight.net/.default'
#Authority
authority = 'https://login.microsoftonline.com/' + tenant_id
app = msal.ConfidentialClientApplication(
client_id , client_secret, authority,
#cache - For details on how look at this example: https://github.com/Azure-Samples/ms-identity-python-webapp/blob/master/app.py
)
# The pattern to acquire a token looks like this.
result = None
result = app.acquire_token_for_client(scopes=[scope])
accessToken = result['access_token']
verify_https = True
request_timeout = 10
# Print access token
print("Access token: " + accessToken)
# API format
api_version = 'v1'
api_format = kafkarest_endpoint + '/{api_version}/{rest_api}'
get_topic_api = 'metadata/topics'
topic_api_format = 'topics/{topic_name}'
producer_api_format = 'producer/topics/{topic_name}'
consumer_api_format = 'consumer/topics/{topic_name}/partitions/{partition_id}/offsets/{offset}?count={count}' # by default count = 1
partitions_api_format = 'metadata/topics/{topic_name}/partitions'
partition_api_format = 'metadata/topics/{topic_name}/partitions/{partition_id}'
# Request header
headers = {
'Authorization': 'Bearer ' + accessToken,
'Content-type': 'application/json' # set Content-type to 'application/json'
}
# New topic
new_topic = 'hello_topic_' + get_random_string()
print("Topic " + new_topic + " is going to be used for demo.")
topics = []
# Create a new topic
# Example of topic config
topic_config = {
"partition_count": 1,
"replication_factor": 1,
"topic_properties": {
"retention.ms": 604800000,
"min.insync.replicas": "1"
}
}
create_topic_url = api_format.format(api_version=api_version, rest_api=topic_api_format.format(topic_name=new_topic))
response = requests.put(create_topic_url, headers=headers, json=topic_config, timeout=request_timeout, verify=verify_https)
print(response.content)
if response.ok:
while new_topic not in topics:
print("The new topic " + new_topic + " is not visible yet. sleep 30 seconds...")
time.sleep(30)
# List Topic
get_topic_url = api_format.format(api_version=api_version, rest_api=get_topic_api)
response = requests.get(get_topic_url, headers={'Authorization': 'Bearer ' + accessToken}, timeout=request_timeout, verify=verify_https)
topic_list = response.json()
topics = topic_list.get("topics", [])
else:
print("Topic " + new_topic + " was created. Exit.")
sys.exit(1)
# Produce messages to new_topic
# Example payload of Producer REST API
payload_json = {
"records": [
{
"key": "key1",
"value": "**********" # A string
},
{
"partition": 0,
"value": 5 # An integer
},
{
"value": 3.14 # A floating number
},
{
"value": { # A JSON object
"id": 1,
"name": "HDInsight Kafka REST proxy"
}
},
{
"value": [ # A list of JSON objects
{
"id": 1,
"name": "HDInsight Kafka REST proxy 1"
},
{
"id": 2,
"name": "HDInsight Kafka REST proxy 2"
},
{
"id": 3,
"name": "HDInsight Kafka REST proxy 3"
}
]
},
{
"value": { # A nested JSON object
"group id": 1,
"HDI Kafka REST": {
"id": 1,
"name": "HDInsight Kafka REST proxy 1"
},
"HDI Kafka REST server info": {
"id": 1,
"name": "HDInsight Kafka REST proxy 1",
"servers": [
{
"server id": 1,
"server name": "HDInsight Kafka REST proxy server 1"
},
{
"server id": 2,
"server name": "HDInsight Kafka REST proxy server 2"
},
{
"server id": 3,
"server name": "HDInsight Kafka REST proxy server 3"
}
]
}
}
}
]
}
print("Payloads in a Producer request: \n", payload_json)
producer_url = api_format.format(api_version=api_version, rest_api=producer_api_format.format(topic_name=new_topic))
response = requests.post(producer_url, headers=headers, json=payload_json, timeout=request_timeout, verify=verify_https)
print(response.content)
# Consume messages from the topic
partition_id = 0
offset = 0
count = 2
while True:
consumer_url = api_format.format(api_version=api_version, rest_api=consumer_api_format.format(topic_name=new_topic, partition_id=partition_id, offset=offset, count=count))
print("Consuming " + str(count) + " messages from offset " + str(offset))
response = requests.get(consumer_url, headers=headers, timeout=request_timeout, verify=verify_https)
if response.ok:
messages = response.json()
print("Consumed messages: \n" + json.dumps(messages, indent=2))
next_offset = response.headers.get("NextOffset")
if offset == next_offset or not messages.get("records", []):
print("Consumer caught up with producer. Exit for now...")
break
offset = next_offset
else:
print("Error " + str(response.status_code))
break
# List partitions
get_partitions_url = api_format.format(api_version=api_version, rest_api=partitions_api_format.format(topic_name=new_topic))
print("Fetching partitions from " + get_partitions_url)
response = requests.get(get_partitions_url, headers={'Authorization': 'Bearer ' + accessToken}, timeout=request_timeout, verify=verify_https)
partition_list = response.json()
print("Partition list: \n" + json.dumps(partition_list, indent=2))
# List a partition
get_partition_url = api_format.format(api_version=api_version, rest_api=partition_api_format.format(topic_name=new_topic, partition_id=partition_id))
print("Fetching metadata of a partition from " + get_partition_url)
response = requests.get(get_partition_url, headers={'Authorization': 'Bearer ' + accessToken}, timeout=request_timeout, verify=verify_https)
partition = response.json()
print("Partition metadata: \n" + json.dumps(partition, indent=2))
Ниже приведен еще один пример того, как можно получить токен из Azure для прокси-сервера REST с помощью команды curl. Обратите внимание, что при получении токена нужно указать scope=https://hib.azurehdinsight.net/.default.
curl -X POST -H "Content-Type: application/x-www-form-urlencoded" -d 'client_id=<clientid>&client_secret=<clientsecret>&grant_type=client_credentials&scope=https://hib.azurehdinsight.net/.default' 'https://login.microsoftonline.com/<tenantid>/oauth2/v2.0/token'