GitHubじゃ!Pythonじゃ!

GitHubからPython関係の優良リポジトリを探したかったのじゃー、でも英語は出来ないから日本語で読むのじゃー、英語社会世知辛いのじゃー

dpkp

kafka-python – Apache Kafka用のPythonクライアント

投稿日:

Apache Kafka用のPythonクライアント http://kafka-python.readthedocs.io/

Kafka Pythonクライアント

Apache Kafka分散ストリーム処理システム用のPythonクライアント。 kafka-pythonは公式のJavaクライアントのように機能し、pythonicインタフェース(消費者イテレータなど)が散在するように設計されています。

kafka-pythonは、新しいブローカー(0.9+)で最もよく使用されますが、古いバージョン(0.8.0まで)と下位互換性があります。 一部の機能は、新しいブローカーでのみ有効になります。 たとえば、完全に調整されたコンシューマ・グループ(つまり、同じグループ内の複数のコンシューマへの動的パーティション割当て)では、0.9+カフカ・ブローカが必要です。 初期のブローカーリリースでこの機能をサポートするには、カスタムリーダーシップの選挙とメンバーシップ/健康診断コードを書いて維持する必要があります(おそらく、動物園または領事を使用して)。 古いブローカの場合、シェフ、アシスタントなどの設定管理ツールを使用して、各コンシューマインスタンスに異なるパーティションを手動で割り当てることで、同様のことを実現できます。このアプローチは正常に動作しますが、障害時のリバランスをサポートしません。 詳細は、< https://kafka-python.readthedocs.io/ja/master/compatibility.html >を参照してください。

マスターブランチにはリリースされていない機能が含まれている可能性があることに注意してください。 リリースのドキュメントについては、readthedocsやPythonのインラインヘルプを参照してください。

>>> pip install kafka-python

カフカコンシューマー

KafkaConsumerは、公式のJavaクライアントと可能な限り同じように操作することを目的とした、高度なメッセージコンシューマです。 コーディネートされた消費者グループを完全にサポートするには、Group APIをサポートするkafkaブローカー(kafka v0.9 +)を使用する必要があります。

APIと設定の詳細については、< https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html >を参照してください。

コンシューマイテレータはConsumerRecordsを返します。ConsumerRecordsは、トピック、パーティション、オフセット、キー、および値の基本メッセージ属性を公開する単純な名前付きの組です。

>>> from kafka import KafkaConsumer
>>> consumer = KafkaConsumer('my_favorite_topic')
>>> for msg in consumer:
...     print (msg)
>>> # join a consumer group for dynamic partition assignment and offset commits
>>> from kafka import KafkaConsumer
>>> consumer = KafkaConsumer('my_favorite_topic', group_id='my_favorite_group')
>>> for msg in consumer:
...     print (msg)
>>> # manually assign the partition list for the consumer
>>> from kafka import TopicPartition
>>> consumer = KafkaConsumer(bootstrap_servers='localhost:1234')
>>> consumer.assign([TopicPartition('foobar', 2)])
>>> msg = next(consumer)
>>> # Deserialize msgpack-encoded values
>>> consumer = KafkaConsumer(value_deserializer=msgpack.loads)
>>> consumer.subscribe(['msgpackfoo'])
>>> for msg in consumer:
...     assert isinstance(msg.value, dict)
>>> # Get consumer metrics
>>> metrics = consumer.metrics()

KafkaProducer

KafkaProducerは高レベルの非同期メッセージプロデューサです。 このクラスは、公式のJavaクライアントに可能な限り同様に動作するように意図されています。 詳細は、< https://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.html >を参照してください。

>>> from kafka import KafkaProducer
>>> producer = KafkaProducer(bootstrap_servers='localhost:1234')
>>> for _ in range(100):
...     producer.send('foobar', b'some_message_bytes')
>>> # Block until a single message is sent (or timeout)
>>> future = producer.send('foobar', b'another_message')
>>> result = future.get(timeout=60)
>>> # Block until all pending messages are at least put on the network
>>> # NOTE: This does not guarantee delivery or success! It is really
>>> # only useful if you configure internal batching using linger_ms
>>> producer.flush()
>>> # Use a key for hashed-partitioning
>>> producer.send('foobar', key=b'foo', value=b'bar')
>>> # Serialize json messages
>>> import json
>>> producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'))
>>> producer.send('fizzbuzz', {'foo': 'bar'})
>>> # Serialize string keys
>>> producer = KafkaProducer(key_serializer=str.encode)
>>> producer.send('flipflap', key='ping', value=b'1234')
>>> # Compress messages
>>> producer = KafkaProducer(compression_type='gzip')
>>> for i in range(1000):
...     producer.send('foobar', b'msg %d' % i)
>>> # Get producer performance metrics
>>> metrics = producer.metrics()

スレッドの安全性

KafkaProducerは、カフカコンシューマーとは異なり、問題なくスレッド間で使用できます。

スレッドローカル方式でKafkaConsumerを使用することは可能ですが、マルチプロセッシングが推奨されます。

圧縮

kafka-pythonは、gzip圧縮/解凍をネイティブにサポートしています。 圧縮されたlz4メッセージを生成または使用するには、python-lz4(pip install lz4)をインストールする必要があります。 スナッピー圧縮/圧縮解除を有効にするには、python-snappyをインストールします(スナッピーライブラリも必要です)。 詳細については、< https://kafka-python.readthedocs.io/en/master/install.html#optional-snappy-install >を参照してください。

プロトコル

kafka-pythonの第2の目標は、python replを介してkafkaブローカーとやり取りするための使いやすいプロトコル層を提供することです。 これは、テスト、プロービング、および一般的な実験に役立ちます。 プロトコルサポートは、カフカブローカーを調べ、実行中のバージョン(0.8.0〜1.0)を識別するKafkaClient.check_version()メソッドを有効にするために利用されます。

低レベル

レガシーサポートは、低レベルのコンシューマおよびプロデューサクラス、SimpleConsumerおよびSimpleProducerで維持されます。 APIの詳細については、< https://kafka-python.readthedocs.io/en/master/simple.html?highlight=SimpleProducer >を参照してください。







-dpkp

執筆者: