quixio/quix-streams

quixio/quix‑streams – Kafka 上でのリアルタイムデータエンジニアリング用 Python フレームワーク

何であるか – Apache Kafka 上で信頼性の高いエンドツーエンドのストリーミングパイプラインを構築できるオープンソースの Python ライブラリです。Streaming DataFrame API と軽量な Application オブジェクトを提供し、Kafka 接続、シリアル化、ステート管理、正確に一度だけの保証を処理するため、変換ロジックに集中できます。

主な機能

• プア・Python 実装 – Java ラッパーまたはクロスランゲージデバッグの必要なし。
• 宣言的で表形式の処理（apply、filter、group-by、window、join など）に適した Streaming DataFrame API。
• 任意の Kafka トピックに接続可能なプラグイン可能な Sources と Sinks。
• JSON、Avro、Protobuf および Schema Registry 用の組み込みシリアライザ。
• フォールトトレラントなステートフル演算子と自動チェックポイント機能。
• Kafka トランザクションによる正確に一度だけの処理。
• 任意の Python ベースの分析または ML ライブラリと連携可能。ライブストリームにモデル推論やトレーニングを簡単に埋め込める。

一般的なワークフロー

from quixstreams import Application

app = Application(broker_address="localhost:9092")

in_topic  = app.topic("temperature-celsius", value_deserializer="json")
out_topic = app.topic("temperature-alerts",  value_serializer="json")

sdf = app.dataframe(topic=in_topic)
# 変換、フィルタリング、その後出力
sdf = sdf.apply(lambda v: {"temperature_F": v["temperature"]*9/5 + 32})
        .filter(lambda v: v["temperature_F"] > 150)
        .to_topic(out_topic)

app.run()

Application は自動的に消費、デシリアライズ、DataFrame ロジックの実行、結果の生成、ステートのチェックポイントを処理します。

導入方法

• PyPI からインストール: python -m pip install quixstreams（または Conda）。
• Python 3.11+ および Kafka 0.10+ を必要とします。
• 完全なドキュメント、クイックスタートガイド、および複数のチュートリアル（単語カウント、異常検出、購入フィルタリング、太陽光発電所の拡張など）は https://quix.io/docs/quix-streams で公開されています。

実行環境

• Python が利用可能な場所であればどこでも – オンプレミスサーバー、クラウド VM、コンテナ、または管理型の Quix Cloud プラットフォーム（AWS、Azure、GCP）。既存の Kafka ブローカー以外に追加のサーバーサイドクラスタは不要です。

コミュニティとサポート

• Slack コミュニティ、YouTube チャンネル、X（Twitter）、LinkedIn で発表やチュートリアルを配信。
• イシューは GitHub で管理。プロジェクトは Apache 2.0 ライセンスです。

誰が使うか

• リアルタイム ETL/ELT パイプラインを構築するデータエンジニア。
• 推論やオンライン学習のためにモデルとの間でデータをストリーミングする必要がある機械学習チーム。
• Kafka のスケーラビリティと耐久性に依存するイベント駆動型マイクロサービスの開発者。

---

ライセンス – Apache 2.0（LICENSE ファイルを参照）。