Confluent の Python 公式チュートリアルを Databricks で実行してみた


## 概要

Confluent のサイトにて紹介されている`Getting Started with Apache Kafka and Python`をノートブック型 python 環境である Databricks 上で実行する方法を紹介します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d0e1f013-4d4b-9af2-53f6-458a45fc4fc8.png)

引用元：[Apache Kafka and Python - Getting Started Tutorial](https://developer.confluent.io/get-started/python/)

実施手順では左側の目次に合わせて実施内容を説明します。細かい手順は紹介しないため、必要に応じて下記ページを参照してください。

1. [Introduction](https://developer.confluent.io/get-started/python/#introduction)
2. [Prerequisites](https://developer.confluent.io/get-started/python/#prerequisites)
3. [Create Project](https://developer.confluent.io/get-started/python/#create-project)
4. [Kafka Setup](https://developer.confluent.io/get-started/python/#kafka-setup)
5. [Create Topic](https://developer.confluent.io/get-started/python/#create-topic)
6. [Build Producer](https://developer.confluent.io/get-started/python/#build-producer)
7. [Build Consumer](https://developer.confluent.io/get-started/python/#build-consumer)
8. [Produce Events](https://developer.confluent.io/get-started/python/#produce-events)
9. [Consume Events](https://developer.confluent.io/get-started/python/#consume-events)
10. [Where next?](https://developer.confluent.io/get-started/python/#where-next)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c0b89322-5d83-e8f5-9d0b-bde3bdeb2700.png)

引用元：[Introduction](https://developer.confluent.io/get-started/python/#introduction)


## 実施手順

### Introduction

Confluent の無償トライアルに申し込みます。Google アカウントでもうしこんだところ、すぐにアカウントが払い出されました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bcbf95ca-6691-83b1-abcb-655cc723bde0.png)

引用元：[無料版の Confluent をお試しください：あらゆるクラウドでお使いいただけるマネージド型 Kafka です | JP](https://www.confluent.io/ja-jp/get-started/)

### Prerequisites

Databricks 上でシングルノードのクラスターを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/79267d32-a5e6-1124-7398-2faff03b7fe8.png)

### Create Project

Databricks 上で`confluent-kafka`をインストールします。

```python
%pip install confluent-kafka -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/30da0593-c7e9-bc6d-5ce1-bed773605d7a.png)

### Kafka Setup

Confuent にて kafka 環境の作成と認証情報を取得します。 

kafka 環境の作成については、ドキュメント通りの手順を実施しました。クレジットカードの入力が求められた場合には、`CONFLUENTDEV1`というプロモーションコードを入力することでスキップできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c36891d5-fa5f-8592-7ceb-8f481e436cae.png)

引用元：[Apache Kafka and Python - Getting Started Tutorial](https://developer.confluent.io/get-started/python/?_ga=2.176204389.1927867639.1733706324-1213280603.1732583603&_gl=1*1bzx8jq*_gcl_au*NzA0NTY0NzYuMTczMjU4MzYwMy41NzE0NjM3MTAuMTczMzcyNjYzNy4xNzMzNzI2NjM2*_ga*MTIxMzI4MDYwMy4xNzMyNTgzNjAz*_ga_D2D3EGKSGD*MTczMzk2NTY0Ni45LjEuMTczMzk2NTg0Ny4xOS4wLjA.#kafka-setup)

認証情報については`BASIC`の手順を実施しました。APIのキーとシークレットが表示されますが、`Download and continue`を選択してダウンロードしたテキストファイルを参照することをおすすめします。接続先に関する情報を確認できるためです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f8e75d19-3e09-cae8-ddc3-a210f860c624.png)

引用元：[Apache Kafka and Python - Getting Started Tutorial](https://developer.confluent.io/get-started/python/#kafka-setup)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/178b5dad-3b83-7906-c54a-d0918b6f05e8.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/51b0ffa9-15f7-39d5-4fa5-579af0006b09.png)


Databricks のノートブック上で、接続情報と認証情報を変数にセットします。シークレットなどの機微なデータは、通常であれば Databricks シークレットに格納しますが、動作確認であるため平文でセットしています。

```python
bootstrap_servers = ""
sasl_username = ""
sasl_password = ""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/29b555ee-1a45-a731-ac3f-30b09a3a6742.png)

### Create Topic

Confluet 上で`purchases`という topic を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2b85199c-3f96-0574-fb26-e810b5b5b6ba.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/73df7501-f712-6444-4272-807ff25edbdc.png)

### Build Producer

Databricks ノートブックに、ノートブック上で動作することを考慮した修正済みコードを記述します。

```python
#!/usr/bin/env python
from random import choice
from confluent_kafka import Producer


config = {
    # User-specific properties that you must set
    'bootstrap.servers': bootstrap_servers,
    'sasl.username':     sasl_username,
    'sasl.password':     sasl_password,

    # Fixed properties
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms':   'PLAIN',
    'acks':              'all'
}

# Create Producer instance
producer = Producer(config)

# Optional per-message delivery callback (triggered by poll() or flush())
# when a message has been successfully delivered or permanently
# failed delivery (after retries).
def delivery_callback(err, msg):
    if err:
        print('ERROR: Message failed delivery: {}'.format(err))
    else:
        print("Produced event to topic {topic}: key = {key:12} value = {value:12}".format(
            topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))

# Produce data by selecting random values from these lists.
topic = "purchases"
user_ids = ['eabara', 'jsmith', 'sgarcia', 'jbernard', 'htanaka', 'awalther']
products = ['book', 'alarm clock', 't-shirts', 'gift card', 'batteries']

count = 0
for _ in range(10):
    user_id = choice(user_ids)
    product = choice(products)
    producer.produce(topic, product, user_id, callback=delivery_callback)
    count += 1

# Block until the messages are sent.
producer.poll(10000)
producer.flush()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/77bf4acf-40b2-ab88-fb97-0208902551d1.png)

記事を書く前の検証時にはコードを修正したのですが、修正しなくても動作はするようです。

```python
if __name__ == '__main__':
    print(__name__)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/51d53241-01d4-cb5b-63d0-5679f335b242.png)

### Build Consumer

Databricks ノートブックに、ノートブック上で動作することを考慮した修正済みコードを記述します。

```python
#!/usr/bin/env python
from confluent_kafka import Consumer


config = {
    # User-specific properties that you must set
    'bootstrap.servers': bootstrap_servers,
    'sasl.username':     sasl_username,
    'sasl.password':     sasl_password,

    # Fixed properties
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms':   'PLAIN',
    'group.id':          'kafka-python-getting-started',
    'auto.offset.reset': 'earliest'
}

# Create Consumer instance
consumer = Consumer(config)

# Subscribe to topic
topic = "purchases"
consumer.subscribe([topic])

# Poll for new messages from Kafka and print them.
try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            # Initial message consumption may take up to
            # `session.timeout.ms` for the consumer group to
            # rebalance and start consuming
            print("Waiting...")
        elif msg.error():
            print("ERROR: %s".format(msg.error()))
        else:
            # Extract the (optional) key and value, and print.
            print("Consumed event from topic {topic}: key = {key:12} value = {value:12}".format(
                topic=msg.topic(), key=msg.key().decode('utf-8'), value=msg.value().decode('utf-8')))
except KeyboardInterrupt:
    pass
finally:
    # Leave group and commit final offsets
    consumer.close()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8c106123-5a7a-0344-7bf7-10bf27f73cc0.png)

### Produce Events

`Build Producer`の手順で記述したコードを実行します。Confluent 上においてもデータが書きこまれたことを確認できます。

```text
Produced event to topic purchases: key = jbernard     value = batteries   
Produced event to topic purchases: key = jbernard     value = book        
Produced event to topic purchases: key = awalther     value = batteries   
Produced event to topic purchases: key = eabara       value = batteries   
Produced event to topic purchases: key = jbernard     value = alarm clock 
Produced event to topic purchases: key = htanaka      value = gift card   
Produced event to topic purchases: key = htanaka      value = book        
Produced event to topic purchases: key = eabara       value = book        
Produced event to topic purchases: key = jsmith       value = book        
Produced event to topic purchases: key = awalther     value = batteries   
0
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c9874bcc-f69f-ba6e-a77c-711d53814092.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9cc68452-4fb4-2d01-fbd3-fadb96ae8cbf.png)

### Consume Events

`Build Consumer`の手順で記述したコードを実行します。処理が継続実行されるため、セルの実行をとめることを忘れないようでしてください。

```text
Consumed event from topic purchases: key = jbernard     value = batteries   
Consumed event from topic purchases: key = jbernard     value = book        
Consumed event from topic purchases: key = awalther     value = batteries   
Consumed event from topic purchases: key = eabara       value = batteries   
Consumed event from topic purchases: key = jbernard     value = alarm clock 
Consumed event from topic purchases: key = htanaka      value = gift card   
Consumed event from topic purchases: key = htanaka      value = book        
Consumed event from topic purchases: key = eabara       value = book        
Consumed event from topic purchases: key = jsmith       value = book        
Consumed event from topic purchases: key = awalther     value = batteries   
Waiting...
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/22cbe64b-31c1-77d4-8fd6-2cbd842434c7.png)

セルを再実行した際には、想定通りに過去に取得したデータを取得しませんでした。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5c03f7df-09cb-6635-f397-dc35080c691b.png)

### Where next?

Confluent のサイトでは無償のトレーニングが動画が公開されているためおすすめです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5fafd412-1471-4387-e102-70ff467c17ee.png)

引用元：[Learn Apache Kafka® & Flink®](https://developer.confluent.io/courses/#fundamentals)

Python 開発者向けのコンテンツもあります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c100a523-bbce-e53b-a378-7c0479049034.png)

引用元：[Introduction to Python for Kafka](https://developer.confluent.io/courses/kafka-python/intro/)