Python

VPC 환경에서 이용 가능합니다.

TCP 접속 엔드포인트를 사용하여 토픽에 저장된 데이터(메시지)를 수신하는 방법과 예제 코드를 설명합니다.

주의

보안을 위해 데이터 전송 시 메인 계정의 인증 정보를 사용할 수 없으므로 반드시 서브 계정의 인증 정보를 사용해 주십시오.

사전 준비(인증 정보 확인 및 SASL 설정)

Kafka 클라이언트 애플리케이션을 사용하여 Data Stream 서비스에 접속하려면 Kafka 클라이언트 종류에 맞게 SASL 설정을 추가해야 합니다. Sub Account 인증 정보 확인 및 SASL 설정 방법은 다음과 같습니다.

Data Stream 서비스에 접속할 사용자의 Sub Account 인증 정보(Access key와 Secret key)를 확인해 주십시오.
- Sub Account 인증 정보는 SASL 인증 설정 시 사용자명(username)과 비밀번호(password)로 사용됩니다.
- 서브 계정의 Access Key 발급 및 확인: 서브 계정 생성 참조

Kafka 클라이언트 종류(Python, Java, Logstash)에 맞게 데이터 송·수신 코드에 SASL 설정을 추가해 주십시오.

Python, Java

# username: Sub Account Access Key 입력
# password: Sub Account Secret Key 입력

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="{ncp_iam_SubAccountAccessKey}" password="{ncp_iam_SubAccountSecretKey}";

데이터 송신 예제

Data Stream 서비스는 Kafka Client를 지원합니다. Kafka Client 라이브러리를 활용하여 토픽에 데이터를 송신할 수 있습니다. 데이터 송신 예제를 소개합니다.

Python producer

Python 예제 코드를 사용하여 데이터를 송신하는 방법은 다음과 같습니다.

다음 명령어로 Python Kafka 클라이언트 라이브러리를 설치해 주십시오.
```
pip install kafka-python
```

다음 예시와 같이 사전 준비에서 확인한 Sub Account 및 SASL 인증 정보를 포함하여 Python 코드를 작성한 후 실행해 주십시오.

import ssl  

from kafka import KafkaProducer  
import json  

brokers = "{TOPIC_접속_엔드포인트}"  
topic_name = "{TOPIC_NAME}"  

username = "{ncp_iam_SubAccountAccessKey}"  
password = "{ncp_iam_SubAccountSecretKey}"

security_protocol = 'SASL_SSL'  
sasl_mechanism = 'PLAIN'  

# Create a new context using system defaults, disable all but TLS1.2  
context = ssl.create_default_context()  
context.options &= ssl.OP_NO_TLSv1  
context.options &= ssl.OP_NO_TLSv1_1  

producer = KafkaProducer(bootstrap_servers = brokers,  
                        sasl_plain_username = username,  
                        sasl_plain_password = password,  
                        security_protocol = security_protocol,  
                        ssl_context = context,  
                        sasl_mechanism = sasl_mechanism,  
                        api_version = (0,10),  
                        retries=5)  


partition = 0
key = "key-1"
value_dict = {'key':  key, 'data': "this is a test (body)"}  
msg_value_json = json.dumps(value_dict)

producer.send(topic=topic_name,  
            key=bytes(key, 'utf-8'),  
            value=msg_value_json.encode("utf-8"),  
            partition=partition)  

producer.flush()

Python producer with Schema Registry

스키마 레지스트리를 연동하는 경우, Python 예제 코드를 사용하여 데이터를 송신하는 방법은 다음과 같습니다.

스키마 레지스트리를 연동할 때, 다음과 같은 라이브러리가 필요할 수 있습니다.

# Kafka 클라이언트 및 Schema Registry 지원
confluent-kafka[avro]==2.3.0
confluent-kafka[schema-registry]==2.3.0

# Avro 직렬화 지원
avro-python3==1.11.3
fastavro==1.9.4

다음 예시와 같이 사전 준비에서 확인한 Sub Account 및 SASL 인증 정보를 포함하여 Python 코드를 작성한 후 실행해 주십시오.

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField
import time

# NCP Connection Settings
brokers = "{TOPIC_접속_엔드포인트}"
topic_name = "{TOPIC_NAME}"
username = "{ncp_iam_SubAccountAccessKey}"
password = "{ncp_iam_SubAccountSecretKey}"
schema_registry_url = "{SCHEMA_REGISTRY_URL}"

# Schema Registry Client
sr_config = {
    'url': schema_registry_url,
    'basic.auth.user.info': f'{username}:{password}'
}
schema_registry_client = SchemaRegistryClient(sr_config)

# User Schema
user_schema_str = """{
  "type": "record",
  "name": "User",
  "namespace": "com.example.kafka",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"},
    {"name": "age", "type": ["null", "int"], "default": null}
  ]
}"""

avro_serializer = AvroSerializer(schema_registry_client, user_schema_str)

# Kafka Producer
producer_config = {
    'bootstrap.servers': brokers,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'PLAIN',
    'sasl.username': username,
    'sasl.password': password,
}
producer = Producer(producer_config)

# Sample data
users = [
    {"id": 1, "name": "John Doe", "email": "john@example.com", "age": 30},
    {"id": 2, "name": "Jane Smith", "email": "jane@example.com", "age": 25},
    {"id": 3, "name": "Bob Johnson", "email": "bob@example.com", "age": None}
]

print(f"Starting to send {len(users)} messages to {topic_name}")

try:
    for user in users:
        # Serialize and send message
        context = SerializationContext(topic_name, MessageField.VALUE)
        serialized_value = avro_serializer(user, context)
        producer.produce(
            topic=topic_name,
            key=f"user-{user['id']}",
            value=serialized_value
        )
        print(f"Sent: {user['name']} (ID: {user['id']})")
    
    producer.flush()
    print("All messages sent successfully!")
    
except Exception as e:
    print(f"Error: {e}")
finally:
    producer.flush()

데이터 수신 예제

Data Stream 서비스는 Kafka Client를 지원합니다. Kafka Client 라이브러리를 활용하여 토픽에 데이터를 수신할 수 있습니다. 데이터 수신 예제를 소개합니다.

Python consumer

Python 예제 코드를 사용하여 데이터를 수신하는 방법은 다음과 같습니다.

다음 명령어로 Python Kafka 클라이언트 라이브러리를 설치해 주십시오.
```
pip install kafka-python
```

다음 예시와 같이 사전 준비에서 확인한 Sub Account 인증 정보를 포함하여 Python 코드를 작성한 후 실행해 주십시오.

import ssl  
from kafka import KafkaConsumer  

brokers = "{TOPIC_접속_엔드포인트}"  
topic_name = "{TOPIC_NAME}"    
username = "{ncp_iam_SubAccountAccessKey}"  
password = "{ncp_iam_SubAccountSecretKey}"

security_protocol = 'SASL_SSL'  
sasl_mechanism = 'PLAIN'  


# Create a new context using system defaults, disable all but TLS1.2  
context = ssl.create_default_context()  
context.options &= ssl.OP_NO_TLSv1  
context.options &= ssl.OP_NO_TLSv1_1  

consumer = KafkaConsumer(topic_name, bootstrap_servers = brokers,  
						group_id = "my-consumer-group-name",  
						auto_offset_reset='earliest',  
						sasl_plain_username = username,  
						sasl_plain_password = password,  
						security_protocol = security_protocol,  
						ssl_context = context,  
						sasl_mechanism = sasl_mechanism)  

from datetime import datetime  
for message in consumer:  
print(f'({datetime.now()}) Topic : {message.topic}, Partition : {message.partition}, Offset : {message.offset}, Key : {message.key}, value : {message.value}')

Python consumer with Schema Registry