本稿では、DataStaxの以下のリポジトリで公開されているPulsar Tansformationsについて紹介します。

概要

Pulsar Tansformationsの目的は、Pulsarファンクションのユースケースにおいて、一般的に行われる変換を実装するパルサー関数を提供し、それらの組み合わせをノーコードで編成できるようにすることです。
Pulsar Transformationsを利用するために、Pulsar スキーマを理解したり、Pulsar Functions でサポートされている言語でのコーディングを理解している必要はありません。

また、複雑なソフトウェアの組み合わせのセットアップする手間を取らせずに、Pulsar Transformationsを Pulsar クラスターで簡単に使用できる環境を提供することも目標としています。

Pulsar Transformationsでは、汎用性の高い基本的な変換の提供を目的としています。
ビジネスロジックの実装のために、Pulsarファンクションを独自に実装する必要性を不要にするものではありません。

現在利用可能な変換は次のとおりです。

cast : キーまたは値のスキーマをターゲットと互換性のあるスキーマに変更します。
drop-fields : 構造化データからフィールドを削除します。
merge-key-value : キーと値の両方が同じスキーマタイプの構造化データである KeyValue レコードのフィールドをマージします。
unwrap-key-value : レコードが KeyValue の場合、KeyValue のキーまたは値を抽出し、それをレコード値にします。
flatten : 構造化データを平坦化します。
drop : レコードを（移行の処理から）削除します。
compute : 新しいプロパティ、値、またはフィールド値を計算します。

Pulsar Transformations を実行するには、Pulsar 2.11 以降または Luna Streaming 2.10 以降が必要です。

コンフィギュレーション

以下にPulsar Transformationsコンフィギュレーションの実例を示します。

{
  "steps": [
    {
      "type": "drop-fields", "fields": "keyField1,keyField2", "part": "key"
    },
    {
      "type": "merge-key-value"
    },
    {
      "type": "unwrap-key-value"
    },
    {
      "type": "cast", "schema-type": "STRING"
    }
  ]
}

steps配列の要素の順に変換が行われます。各ステップの処理はそのtypeによって定義され、そのtypeに応じた引数と組み合わせて利用します。

終わりに

本稿では、Pulsar Transformationsの概要について紹介しました。今後、別稿にて、より具体的な内容を紹介することができればと思っています。

ご興味のある方は、どうぞGithubリポジトリのREADMEや以下のドキュメントをご覧になってください。

また、以下のビデオがStreamNative社のサイトで紹介されています。
https://streamnative.io/videos/pulsar-virtual-summit-europe-2023-build-low-code-stream-data-pipelines-with-pulsar-transformations

上記と同じ著者による以下のブログで、簡潔に全体像を把握することができます。