はじめに
「手を動かして学ぶ！KafkaとPythonによるAWSストリーミングデータ分析入門【30日チャレンジ】」15日目です。これまでの2週間で、私たちはローカルからAWSへと開発環境を移行し、PythonアプリケーションがMSKクラスターで動作することを確認しました。
今日からは、いよいよ 「リアルタイムデータ分析」 というテーマに本格的に取り組みます。今回は、Webサイトのアクセスログを模倣したデータをEC2インスタンス上で生成し、それをKafkaに送信するスクリプトを作成します。これが、ストリーミングデータ分析パイプラインの最初のステップ「データ収集」となります。
1. なぜログデータをKafkaに送信するのか？
多くの企業にとって、Webサイトのアクセスログは非常に貴重な情報源です。しかし、従来のバッチ処理では、以下の課題がありました。
- 分析の遅延: ログファイルを夜間にまとめて処理するため、リアルタイムなユーザーの行動を把握できません。
- システムの複雑さ: ログファイルは増え続けるため、ファイルシステムやデータベースの管理が複雑になります。
Kafkaを使うことで、これらの課題を解決できます。
- リアルタイム処理: ログが生成された瞬間にKafkaに送信し、即座に分析できます。
- スケーラブルなデータハブ: 複数のWebサーバーから生成される膨大なログを、Kafkaが効率的に集約・管理します。
2. EC2上でWebログプロデューサーを実装
今回は、EC2インスタンス上で動かすための、より実践的なプロデューサーのスクリプトを作成します。このスクリプトは、Webサイトのアクセスログを模倣したJSONデータを連続的に生成し、Kafkaに送信します。
EC2インスタンスにSSHで接続し、web_log_producer.pyというファイルを作成して、以下のコードを記述してください。
# web_log_producer.py
from confluent_kafka import Producer
import json
import time
import random
from datetime import datetime
import uuid
# Kafkaブローカーの設定
conf = {'bootstrap.servers': '<your-msk-bootstrap-brokers>'}
# Producerインスタンスの作成
producer = Producer(conf)
# 配信レポートコールバック
def delivery_report(err, msg):
if err is not None:
print(f"Failed to deliver message: {err}")
else:
print(f"Message delivered to {msg.topic()} [{msg.partition()}]")
# 送信先のトピック
topic = "web-logs"
# ログデータを生成して連続的に送信
def generate_log_data():
while True:
user_id = str(uuid.uuid4())
page_id = random.choice(['home', 'product_A', 'product_B', 'about', 'contact'])
ip_address = f"192.168.{random.randint(1, 255)}.{random.randint(1, 255)}"
timestamp = datetime.now().isoformat()
log_data = {
'user_id': user_id,
'timestamp': timestamp,
'event_type': 'page_view',
'page_id': page_id,
'ip_address': ip_address
}
json_data = json.dumps(log_data)
try:
producer.produce(
topic,
key=user_id.encode('utf-8'),
value=json_data.encode('utf-8'),
callback=delivery_report
)
print(f"Sent log: {json_data}")
except Exception as e:
print(f"An error occurred: {e}")
# 1秒間に約5件のログを生成
time.sleep(random.uniform(0.1, 0.3))
# スクリプトの実行
try:
generate_log_data()
except KeyboardInterrupt:
print("\nShutting down producer...")
finally:
producer.flush()
print("All messages flushed. Producer is shut down.")
コードのポイント
-
uuid.uuid4(): ユニークなユーザーIDを生成し、実際の匿名ユーザーを模倣しています。 -
datetime.now().isoformat(): ISO 8601形式でタイムスタンプを生成します。 -
generate_log_data(): 無限ループでログデータを連続的に生成し続けます。これにより、ストリーミングデータの特徴である「継続的なデータの流れ」を再現しています。 -
time.sleep(): データの生成間隔をランダムに設定し、よりリアルなトラフィックを模倣しています。
3. アプリケーションの実行と確認
ステップ1：トピックの作成
このスクリプトを実行する前に、web-logsという新しいトピックをMSKクラスターに作成する必要があります。EC2インスタンス上で以下のコマンドを実行してください。
# MSKのブートストラップブローカーのアドレスを環境変数に設定していると仮定
# BOOTSTRAP_BROKERS=$(aws msk describe-cluster --cluster-arn <your-cluster-arn> --query 'ClusterInfo.BrokerNodeGroupInfo.ClientSubnetInfo.BrokerEndpoints' --output text)
/usr/bin/kafka-topics --create --topic web-logs --bootstrap-server $BOOTSTRAP_BROKERS
ステップ2：プロデューサーの実行
トピックが作成されたら、web_log_producer.pyを実行します。
python3 web_log_producer.py
実行すると、EC2インスタンスのコンソールに、ログデータがKafkaに送信されるたびに、その内容と配信レポートが表示されるはずです。
ステップ3：データの確認
別のSSHセッションでコンシューマーを起動し、データが正常に受信されているか確認しましょう。
# 別のターミナルでEC2に接続
ssh -i "kafka-challenge-key.pem" ec2-user@<your-ec2-public-ip>
# コンソールコンシューマーを起動
/usr/bin/kafka-console-consumer --topic web-logs --bootstrap-server $BOOTSTRAP_BROKERS
コンソールに、プロデューサーが送信したログデータがリアルタイムで表示されれば成功です。
まとめと次回予告
今日は、EC2インスタンス上でWebログデータを生成し、それをMSKクラスターに送信するリアルタイムデータ収集の仕組みを構築しました。これにより、ストリーミングデータ分析パイプラインの最も重要な入り口が完成しました。
明日からは、このKafkaに集められたデータをどう活用していくかを学んでいきます。
16日目: ストリーム処理の基本：フィルタリングと集計を実装
お楽しみに！