<jupyter_start><jupyter_text>このノートブックについて このノートブックは [Confluent Kafka Python] を利用し、Pythonを用いてApache Kafkaクラスタにデータを書き込むサンプルである。[Confluent Kafka Python]: https://docs.confluent.io/ja-jp/clients-confluent-kafka-python/1.5.0/overview.html  準備<jupyter_code>from confluent_kafka import Producer
from confluent_kafka import Consumer
from confluent_kafka import TopicPartition
from confluent_kafka import OFFSET_BEGINNING
from confluent_kafka.admin import AdminClient
from confluent_kafka.admin import NewTopic
import socket
import time<jupyter_output><empty_output><jupyter_text>ここでは簡単化のため、単一ノード上にKafka Broker、ZooKeeper、Clientを起動することとする。そのため、実行ノードの最初にホスト名を取得する。![Pythonクライアント動作確認のためのイメージ](https://github.com/dobachi/PythonKafkaECHONETLiteExample/blob/main/images/ArchitectureOfKafkaBasic.png?raw=true)<jupyter_code>hostname = socket.gethostname()
hostname<jupyter_output><empty_output><jupyter_text>クライアントに渡すコンフィグを生成する。<jupyter_code>producer_conf = {'bootstrap.servers': hostname + ":9092",
                 'client.id': hostname}
consumer_conf = {'bootstrap.servers': hostname + ":9092",
                 'group.id': 0,
                'auto.offset.reset': 'earliest'}<jupyter_output><empty_output><jupyter_text>実験用のトピックを作る<jupyter_code>topic_name = 'test_topic'
num_partition = 1
replication_factor = 1<jupyter_output><empty_output><jupyter_text>コンシューマを作成<jupyter_code>consumer = Consumer(consumer_conf)<jupyter_output><empty_output><jupyter_text>トピックのリストを得るため、メタデータを取得する。<jupyter_code>cluster_metadata = consumer.list_topics()
cluster_metadata<jupyter_output><empty_output><jupyter_text>トピックリストを確認<jupyter_code>cluster_metadata.topics<jupyter_output><empty_output><jupyter_text>`admin` を取得する。<jupyter_code>admin = AdminClient(producer_conf)<jupyter_output><empty_output><jupyter_text>トピックを作成する処理を定義<jupyter_code>def create_topic(admin, topic_name, num_partition, replication_factor):
    new_topic = NewTopic(topic_name, num_partition, replication_factor)
    result = admin.create_topics([new_topic])
    return result<jupyter_output><empty_output><jupyter_text>トピックがあれば削除し、改めて作り直す。もし既存のトピックを流用したい場合は、冒頭の `delete_topics` をしている箇所をコメントアウトすること。<jupyter_code>if topic_name in cluster_metadata.topics:
    admin.delete_topics([topic_name])

result = create_topic(admin, topic_name, num_partition, replication_factor)   

while(not result[topic_name].done()):
    time.sleep(3)

result
consumer.close()<jupyter_output><empty_output><jupyter_text>書き込む 今回は非同期的に書き込む方式を試す。<jupyter_code>producer = Producer(producer_conf)
def acked(err, msg):
    if err is not None:
        print("Failed to deliver message: %s: %s" % (str(msg), str(err)))
    else:
        print("Message produced: %s" % (str(msg)))

producer.produce(topic_name, key="key", value="value", callback=acked)

# Wait up to 1 second for events. Callbacks will be invoked during
# this method call if the message is acknowledged.
producer.poll(1)<jupyter_output>Message produced: <cimpl.Message object at 0x7f8ca7c320e0><jupyter_text>読み込む<jupyter_code>topic_partition = TopicPartition(topic_name, partition=0, offset=OFFSET_BEGINNING)
topic_partition
def my_on_assign(consumer, partitions):
    for p in partitions:
         # some starting offset, or use OFFSET_BEGINNING, et, al.
         # the default offset is STORED which means use committed offsets, and if
         # no committed offsets are available use auto.offset.reset config (default latest)
        p.offset = OFFSET_BEGINNING
    # call assign() to start fetching the given partitions.
    consumer.assign(partitions)
try:
    consumer = Consumer(consumer_conf)
    consumer.subscribe([topic_name], on_assign=my_on_assign)
    
    for i in range(10):
        msg = consumer.poll(timeout=1.0)
        if msg is None: continue
        
        if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write('%% %s [%d] reached end at offset %d\n' %
                                     (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
        else:
            print('%s : %s' % (msg.key(), msg.value()))
finally:
    consumer.close()<jupyter_output>b'key' : b'value'<jupyter_text>後始末 使用したトピックを消す。もし必要な場合はコメントアウトするなどして残すこと。<jupyter_code>admin.delete_topics([topic_name])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10631531122217174, 'fraction_numerical': 0.0068150840527033164, 'mean_word_length': 5.254261363636363, 'pattern_counts': {'":': 2, '<': 52, '<?xml version=': 0, '>': 52, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '455208', 'license': 'Apache-2.0', 'n_tokens_mistral': 1742, 'n_tokens_neox': 1681, 'n_words': 246, 'path': 'BasicExampleOfKafka.ipynb', 'repo_name': 'dobachi/PythonKafkaECHONETLiteExample', 'size': '4402'}