11 ストリーミング

この項では、Oracle AI Data Platformでのストリーミング・データまたは継続的に生成されるデータの使用について説明します。

トピック:

ストリーミングについて

Apache Spark Structured Streaming機能を使用して、Oracle AI Data Platformでストリーミング・データまたは継続的に生成されたデータをほぼリアルタイムで処理できます。

ノートブックとワークフローの両方で、Apache Spark構造化ストリーミングがサポートされています。次のソースおよびシンクは、ストリーム・データの読取り、ストリーム・データの書込みおよびチェックポイントの場所に使用できます。

表11-1サポートされているソースおよびシンク

ソースまたはシンク サポート済?
ボリューム・パス(/Volume/bronze/bucket1) すべての形式でサポートされます
ワークスペース・パス(/Workspace/folder1/) すべての形式でサポートされます
3つの部分名を持つカタログ内の表(catalog.schema.table) Delta形式のみサポート

Parquet、CSV、JSON、ORC形式ではサポートされていません

例1: サポートされているコード

  • streaming_df = spark.readStream.format("delta").table('stdcatalog.stdschema.deltatable')
  • streaming_df.writeStream.format("delta").outputMode("append").option("checkpointLocation", "/Volumes/checkpoints1/").toTable("stdcatalog.stdschema.deltatable")

例2: サポートされていないコード

  • spark.readStream.option("withEventTimeOrder", "true").format("format") .table("stdcatalog.stdschema.samplecsv")
Kafka 3つの部分からなる命名規則のない任意のKafka互換ストリームでサポートされます。

3つの部分からなる命名規則に従い、Kafkaベースのカタログではサポートされていません)

OCIストリーミングサービス サポートされています
OCIオブジェクト・ストレージ・パス(OCI://を使用) 非サポート
Oracle Autonomous AI Lakehouse、Oracle AI Database、Oracle Autonomous AI Transaction Processing ストリーミング(readStreamまたはwriteStream)ではサポートされていません

ノートブックを使用した構造化ストリーミング

Pythonコードを記述して、ノートブック内のストリーム・データを処理できます。ボリューム・パスまたはワークスペース・パスのいずれかがチェックポイントの場所として有効ですが、オブジェクト・ストレージ・パス(oci://形式)はチェックポイントの場所としてサポートされていません。チェックポイントの場所としてボリューム・パスを使用することをお薦めします。


AI Data Platformノートブック・セルでのストリーミング・コードの例


AI Data Platformノートブックでストリーム・データを処理するために使用されるPythonコードの例

ストリーミング・コードの実行中に、ノートブックの「ダッシュボード」タブから、入力レート、処理レート、バッチ期間などのApache Sparkストリーミング関連のイベントを表示できます。


ノートブックの「ダッシュボード」タブが開き、ストリーミング・データが表示されます

コードを段階的に開発しながら、「RAWデータ」タブからRAWストリーミング関連のイベントを表示することもできます。


ストリーミング関連のイベントを表示するノートブックで「Raw Data」タブが開きます

ワークフローを使用したSpark構造化ストリーミングの構成

ストリーム・データを継続的に処理するために、ワークフロー内でストリーミング・タスクを構成できます。

Oracle AI Data Platformでストリーミングを使用したワークフローの使用を開始するには、まずジョブを作成し、そのジョブにノートブックまたはPythonタスクを1つ追加する必要があります。
  1. ワークスペースに移動し、「ワークフロー」をクリックします。
  2. 「Create cluster」アイコン「ジョブの作成」をクリックします。
  3. ジョブの名前と説明を指定します。
  4. 「参照」をクリックし、ジョブを保存する場所をAIデータ・プラットフォームで選択します。「選択」をクリックします。
  5. 「最大同時実行数」に1と入力します。
  6. 「作成」をクリックします。
  7. 作成したジョブをクリックします。
  8. 「タスクの追加」をクリックします。
  9. タスクに名前を指定します。
  10. 「タスク・タイプ」に「ノートブック」または「Python」を選択します。
  11. 「参照」をクリックし、ストリーミング・タスクとして追加するノートブックまたはPythonスクリプトに移動します。「選択」をクリックします。
  12. ノートブックまたはPythonタスクのコンピュート・クラスタ(まだアタッチされていない場合)を選択します。
  13. 「ストリーミング」チェック・ボックスを選択します。「ストリーミング」を選択すると、実行タイムアウトおよびタスク依存関係がオプションとして無効になります。

    「Streaming」チェック・ボックスが選択された状態で「Create Task Details」ページが開きます

  14. 失敗時にタスクが試行する再試行回数を選択します。0より大きい値を選択した場合、ジョブ実行が再試行までに待機する時間、およびタイムアウト時に再試行を試みるかどうかも指定する必要があります。

    再試行回数が1以上の場合のタスク再試行オプション

  15. 「即時実行」をクリックします。
ストリーミング・タスクが開始されると、手動で停止するまで実行が続行されます。定期的な月次メンテナンス中、ストリーミング・タスクはサービスによって停止および再起動され、ユーザー側からのアクションは必要ありません。