Ubuntu distributions can typically use the default /usr/lib/ssl/certs/ca-certificates.crt; MacOS users that installed openssl via Homebrew can typically use /usr/local/etc/openssl/cert.pem.
azure-event-hubs-for-kafka/quickstart/python at master · Azure/azure-event-hubs-for-kafka (github.com)

bootstrap_servers: eventhub namespace 정보이다.

sasl_password: 개별 eventhub의 endpoint connection string이다.

개별 event hub를 생성하려면, 이렇게 eventhub에서 add를 수행하고

생성된 eventhub에서 “shared access policies”를 추가하고,

생성된 ploicy를 열면 Endpoint connection string이 있다. 이것을 추가한다.

끝으로, topic_name은 eventhub의 이름을 넣는다.

Kafka clinet 설치 및 python package 설치

setup.sh를 실행해 구성한다. 이때, conda env를 이용할 경우 pip 설치시 sudo를 제거하고 수행한다.

azure-eventhub-kafka-python/setup.sh at main · CloudBreadPaPa/azure-eventhub-kafka-python (github.com)

Kafka 클라이언트 종류와 차이

참고자료 : Getting started with Apache Kafka in Python | by Adnan Siddiqi | Towards Data Science

위의 문서에 따르면, 세가지 정도의 방법이 있다. Confluent kafka가 성능면에서 우세한 것으로 보이나, 깊이있는 차이가 필요하면 조금 더 확인할 필요가 있다.

Accessing Kafka in Python
There are multiple Python libraries available for usage:

Kafka-Python — An open-source community-based library.
PyKafka — This library is maintained by Parsly and it’s claimed to be a Pythonic API. Unlike Kafka-Python you can’t create dynamic topics.
Confluent Python Kafka:- It is offered by Confluent as a thin wrapper around librdkafka, hence it’s performance is better than the two.

실행을 위한 환경변수 설정

github에 올려둔 producer.py 코드는 환경변수를 이용하도록 작업했다.

azure-eventhub-kafka-python/producer.py at main · CloudBreadPaPa/azure-eventhub-kafka-python (github.com)

이를 위해 sh 파일을 생성해 환경 변수를 넣거나, bash profile 등에 넣어 처리 가능하다. 이런 식으로 만들어서 실행했다.

export ssl_ca_location="/usr/lib/ssl/certs/ca-certificates.crt" # ubuntu ca certificate

export bootstrap_servers="<YOUR-EVENTHUB-NS>.servicebus.windows.net:9093"

export sasl_password="<YOUR-EVENTHUB-CONNECTION-STRING>"

export topic_name="<YOUR-EVENTHUB-NAME>"

python producer.py

실행 후 발생하는 오류

ca-certificate.crt 위치가 잘못되었거나 존재하지 않는 오류 발생했다.

다시 체크해 처리했으며 Ubuntu의 경우 “/usr/lib/ssl/certs/ca-certificates.crt” 위치에 존재한다.

%3|1631670594.732|SSL|python-example-producer#producer-1| [thrd:app]: error:02001002:system library:fopen:No such file or directory: fopen('/path/to/ca-certificate.crt','r')

%3|1631670594.732|SSL|python-example-producer#producer-1| [thrd:app]: error:2006D080:BIO routines:BIO_new_file:no such file

Traceback (most recent call last):

File "producer.py", line 33, in <module>

p = Producer(**conf)

cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create producer: ssl.ca.location failed: error:0B084002:x509 certificate routines:X509_load_cert_crl_file:system lib"}

Azure Event hub의 SKU가 낮아 발생하는 오류. standard 이상으로 올리면 해결된다.

% Waiting for 1 deliveries

%3|1631671736.190|FAIL|python-example-producer#producer-1| [thrd:sasl_ssl://dw-evthub-kafka.servicebus.windows.net:9093/bootstra]: sasl_ssl://dw-evthub-kafka.servicebus.windows.net:9093/bootstrap: SASL authentication error: Kafka protocol is supported for Standard, Premium and Dedicated SKU only. Namespace: dw-evthub-kafka. (after 17ms in state AUTH_REQ)

Topic의 이름을 잘못 지정할 경우 발생하는 오류. Event hub의 이름으로 topic을 지정하면 해결된다.

% Waiting for 1 deliveries

% Message failed delivery: KafkaError{code=UNKNOWN_TOPIC_OR_PART,val=3,str="Broker: Unknown topic or partition"}