ストリームへのデータプッシュ

Foundry でのほとんどのデータ取り込みは、ソースシステムからデータをプルしてデータセットやストリームに同期することに集中しています。

ストリームにおいて、Foundry はイベントベースのワークフローをサポートするための プッシュベースの取り込み をサポートしています。

Foundry でのプッシュベースのレコード取り込みは、一般的な REST サービスと同じ原則に従います。一連の REST エンドポイントを通じて、レコードを消費し、それらをストリームとデータセットに書き込むことができるプッシュベースの API を公開しています。ストリームにプッシュするために必要な高レベルの情報は次のとおりです：

ストリームのデータセットリソース識別子。
ブランチの名前。
リクエストを認証するためのトークン。

すでにデータの Foundry ソースが設定されている場合は、そのソースに接続することをお勧めします。既存のソースとのストリーミング同期を設定する方法、または Kafka ソースを設定する方法を学びましょう。

以下では、ストリームにデータをプッシュするために必要な手順について説明します：

新しいストリームを設定します。
ストリームにレコードをプッシュします。
ストリームを共有します。
ストリームをテストします。

パート 1. 初期設定

まず、ストリーム作成ワークフローを始めます。

Foundry にログインします。
プロジェクトに移動します。
右上の角にある + New を選択します。
下にスクロールして Stream を選択します。

次に、ストリームのスキーマ、スループット、キーを定義する必要があります。

スキーマ はストリームデータの構造と型を定義します。
スループット はデータ処理率を表し、ストリームで使用されるパーティションの数に影響します。スループットとパーティションのドキュメンテーションを参照してください。
キーは複数のパーティションを使用する際にユニークIDの順序を保証するために使用されます。ストリーミングキーのドキュメンテーションを参照してください。

このチュートリアルでは、シンプルなシングルパーティションのストリームを作成します。

スキーマを sensor_id: String, temperature: Double, created_at: Timestamp に設定します。
スループットを Normal に設定します。

次に、ストリームを設定します。

既存の JSON ブロブから自動的にスキーマを生成するには、Generate from JSON sample... ボタンを選択し、既存の JSON ブロブを貼り付けます。
関係データベースからリアルタイムの更新をストリーミングする場合は、変更データキャプチャを設定します。
ストリームの並列性要件を考慮して、高スループット設定が必要かどうかを判断します。詳細はパーティションのドキュメンテーションを参照してください。
複数のパーティションを使用する際に順序の保証のためのキーを設定することを検討してください。

すべての検証エラーは、Create stream を選択する前に対処する必要があります。ページの下部のツールチップにマウスを置くと、エラーの詳細が表示されます。

右下の角にある Create stream ボタンを選択して Connect ページに移動します。ここでは、ストリーミングデータに接続する方法を指定できます。

パート 2. ストリームへのレコードのプッシュ

これでストリームに接続する準備が整いました。

プッシュベースの取り込みの場合、Connect via API セクションのオプションの一つを選択します。

HTTP リクエストを作成できる任意の言語や技術を使用してレコードをプッシュすることができます。cURL、Python、Node 経由の JavaScript、Java の例を提供しています。Foundry の同期を介してデータを取り込むのではなく、データを Foundry にプッシュしたい場合は、ストリーミング同期の設定方法を確認してくださいストリーミング同期の設定。

この例では、 cURL を選択します。

すると、Push データワークフローページに移動します。
次に、認証メカニズムを選択します。リクエストへの認証は2つの方法をサポートしています：
- サードパーティアプリケーションを使用したプッシュ（推奨）： この方法は OAuth2 ワークフローを使用して、ストリームにレコードをプッシュするために使用できる安全なトークンを作成します。
- パーソナルトークンを使用したプッシュ： この方法はテスト目的のみのユーザー生成トークンを使用します。
このチュートリアルでは、サードパーティアプリケーションを使用したプッシュ 方法を選択します。画面上の手順に従ってサードパーティアプリケーションを設定し、クライアントシークレットを作成します。
今度は、Go to third-party applications を選択して、Foundry プラットフォーム設定のサードパーティアプリケーション管理ページを開きます。
画面右上の Register new application を選択します。
名前を選択し、クライアントのタイプを Server application に設定します。
Create を選択すると、クライアント ID とシークレットが表示されます。

このページから離れると、クライアントシークレットはアクセスできなくなります。安全な場所に保存してください。
これで、クライアント ID とシークレットを Push ワークフローページに追加できます。
ワークフローの Configuring the Application セクションまで下にスクロールし、Manage application を選択して、Foundry プラットフォーム設定の Third-party applications 管理ページを開きます。
次に、Client credentials grant 設定を有効にします。
最後に、右上の角にある Save をクリックします。

パート 3: ストリームの共有

次に、作成したアプリケーションとストリームを共有する必要があります。

最初に、Push ワークフローページに戻ります。ワークフローの Using Your New Application セクションで、生成したクライアント ID を見つけます。
次に、右上の Share を選択して、ストリーム Details サイドバーの Roles タブを開きます。
クライアント ID を Roles 検索フィールドにコピーして貼り付け、作成したアプリケーションを見つけます。 + を選択して Editor ロールを検索し、選択します。
サイドパネルの下部にある Save を選択してストリームを共有します。
プッシュワークフローに戻り、Next step を選択します。

パート 4: ストリームのテスト

ストリームにテストレコードをプッシュするために使用できるコード例が表示されます。コード例は、前の手順で選択した言語によって異なります。この例では、cURL を使用しています。

cURL コマンドを実行する前に、コマンドラインの JSON パースのために jQuery をインストールしておいてください。

最初に、最初のボックスからコマンドをコピーします。これは Foundry の OAuth2 エンドポイントをヒットし、レコードをプッシュするために使用できるアクセストークンを提供します。

Mac、Windows、または Linux マシンの bash ターミナルでコマンドを実行します。

コマンドを実行すると、$ACCESS_TOKEN という名前の使用可能な変数が得られます。この変数を次のコマンドでレコードをプッシュするために使用します。

2つ目のコマンドは、cURL を使用して Foundry エンドポイントをヒットし、ストリームに挿入したいレコードを含むポストリクエストを行います。コマンドは、ストリームにプッシュするダミーレコードで事前に入力されていますが、ストリームのスキーマに準拠した任意のデータを HTTP リクエストに提供できます。
2つ目のコマンドをターミナルにコピーして貼り付け、実行します。

コマンドが成功すると、ストリームにレコードが表示されます。

ストリームにプッシュされるデータを変更するには、ポストリクエストのデータパラメータを修正します。

[
  { 
    \"value\": {
      \"sensor_id\":\"sensor4\",  // センサーのIDを表す
      \"temperature\":4.132  // 温度を表す (摂氏)
    } 
  }
]