stream-processing

ストリーム処理

目的

このスキルを使うと、Kafka、Flink、Apache Spark などのフレームワークを使用して、継続的なデータストリームをリアルタイムで処理できます。データエンジニアリングのパイプラインを支えるため、データの即時取り込み、変換、分析が必要なシナリオ向けに設計されています。

使用する場面

リアルタイム分析が必要な IoT センサー、ログファイル、金融取引など、大量のデータソースに対してこのスキルを使用します。システムメトリクスの監視、異常の検出、ダッシュボードの動的な更新など、バッチ処理では不十分な場合に適用してください。

主な機能

• Kafka の分散アーキテクチャを使って高スループットのストリームを処理し、フォールトトレランスのためにトピック、パーティション、レプリケーションをサポートします。
• Flink でウィンドウ処理（例：1 分間の集計に使うタンブリングウィンドウ）と exactly-once 処理セマンティクスを利用して、状態を持つ計算を実行します。
• Apache Spark Streaming を統合し、DStreams または Structured Streaming API を活用して、map や reduce などの変換をスケーラブルに処理します。
• Flink の設定可能なチェックポイント間隔などを利用したバックプレッシャー処理をサポートし、過負荷を防ぎます。

使用パターン

• プロデューサー・コンシューマーパターン：Kafka プロデューサーを通じてデータを取り込み、Flink コンシューマーで処理します。たとえば、ログを Kafka トピックに送信し、Flink でリアルタイムにフィルタリングおよび集計します。
• ウィンドウ集計：Flink で時間ベースのウィンドウを適用してデータを要約します。たとえば、1 分ごとにイベント数を数えます。
• ETL パイプライン：Spark Streaming を使って Kafka から抽出し、SQL クエリで変換して、Elasticsearch などのデータベースにロードします。
• フォールトトレラント処理：Flink ジョブでチェックポイントを設定し、障害発生後に処理を再開できるようにして、本番環境でのデータ損失を防ぎます。

一般的なコマンド／API

• Kafka CLI コマンド：kafka-console-producer --topic my-topic --broker-list localhost:9092 を使ってメッセージを送信します。メッセージを受信するには、kafka-console-consumer --topic my-topic --from-beginning --bootstrap-server localhost:9092 を使用します。
• Flink コマンド：flink run -c com.example.StreamJob /path/to/jar --input kafka-topic --output file:///output を使ってジョブを送信し、ストリームを処理します。監視には Flink の REST API（http://localhost:8081/jobs/overview）を使用します。
• Spark Streaming API：Scala では、val stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").load() を使ってストリームを作成します。次に、変換を適用します：stream.selectExpr("CAST(value AS STRING)").writeStream.outputMode("append").format("console").start()。
• 設定形式：Kafka では、プロデューサー向けに key.serializer=org.apache.kafka.common.serialization.StringSerializer のようなプロパティファイルが必要です。Flink では設定に YAML を使用します。例：execution.checkpointing.interval: 1min。

統合に関する注意事項

Flink ジョブに依存関係（例：Maven の場合は <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId></dependency>）を追加して、Kafka を Flink のソースとして統合します。認証では、プロデューサースクリプトに $KAFKA_API_KEY のような環境変数を設定します：export KAFKA_API_KEY=your_key; kafka-console-producer --broker-list localhost:9092 --producer.config /path/to/config.properties。Spark に組み込みのコネクターを使って Kafka を接続し、クラスターとの互換性（例：Spark 3.x と Kafka 2.8+）を確保してください。外部サービスには、環境変数を通じて API キーを使用します。たとえば、Spark クラスターへの接続には $SPARK_MASTER_URL を使用します。

エラー処理

プロデューサーにリトライを実装するなどして、Kafka の接続エラーに対処します。たとえば、指数バックオフを使用するループを実装できます：try { producer.send(record) } catch (Exception e) { Thread.sleep(2000 * attempts); }。Flink では、env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS))) を設定して再起動戦略を有効にし、タスクの失敗から復旧できるようにします。Spark では、ストリーミングクエリでチェックポイントを使用します：writeStream.option("checkpointLocation", "/path/to/checkpoints").start()。これにより、障害時に状態を復元できます。SLF4J などを使って構造化形式でエラーを記録し、Prometheus などのツールでリアルタイムのアラートを監視します。

具体的な使用例

1. Kafka-Flink によるリアルタイムログ処理：kafka-console-producer --topic logs --broker-list localhost:9092 を実行し、ログを Kafka に取り込みます。次に、Flink ジョブを実行します：flink run -c com.example.LogProcessor /path/to/jar --input logs。このジョブでは、エラーをフィルタリングします：env.addSource(new FlinkKafkaConsumer<>("logs", ...)).filter(line -> line.contains("ERROR")).print()。
2. Spark Streaming によるセンサーデータの集計：Spark で Kafka からデータを読み込みます：val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "sensors").load()。データを集計します：df.groupBy(window($"timestamp", "1 minute")).avg("value").writeStream.format("console").start()。これにより、IoT センサーのストリームデータを処理し、1 分単位の平均値を算出します。

グラフの関係

• 関連するクラスター：data-engineering
• 関連するスキル：data-ingestion（データソースとして）、machine-learning（ストリーム上でのリアルタイムモデル推論に使用）
• 依存関係：Kafka/Flink クラスターのデプロイには、containerization などのスキルが必要です