ストリームを管理する

次の操作を学べます。

• データ取り込み用のストリームを作成し、設定する
• ストリームの設定を確認し、更新する
• 不要になったストリームを削除する

ストリームは、ストリーム名を使った SQL テーブルとしてパイプラインから利用できます（例: SELECT * FROM my_stream）。

ストリームを作成するには、pipelines streams create コマンドを実行します。

npx wrangler pipelines streams create <STREAM_NAME>

ストリーム、シンク、パイプラインの設定を案内する対話型セットアップウィザードを使う場合は、pipelines setup コマンドを実行します。

npx wrangler pipelines setup

ストリームは、データの扱い方として次の 2 つの方法をサポートします。

• 構造化ストリーム: 特定のフィールドとデータ型でスキーマを定義します。イベントはスキーマに対して検証されます。
• 非構造化ストリーム: 検証なしで、有効な JSON を受け付けます。これらのストリームには、JSON データを格納する value 列が 1 つあります。

構造化ストリームを作成するには、スキーマファイルを指定します。

npx wrangler pipelines streams create my-stream --schema-file schema.json

スキーマファイルの例:

{
	"fields": [
		{
			"name": "user_id",
			"type": "string",
			"required": true
		},
		{
			"name": "amount",
			"type": "float64",
			"required": false
		},
		{
			"name": "tags",
			"type": "list",
			"required": false,
			"items": {
				"type": "string"
			}
		},
		{
			"name": "metadata",
			"type": "struct",
			"required": false,
			"fields": [
				{
					"name": "source",
					"type": "string",
					"required": false
				},
				{
					"name": "priority",
					"type": "int32",
					"required": false
				}
			]
		}
	]
}

サポートするデータ型:

• string - テキスト値
• int32、int64 - 整数
• float32、float64 - 浮動小数点数
• bool - 真偽値（true/false）
• timestamp - RFC 3339 のタイムスタンプ、または単位に応じて Unix 秒、ミリ秒、マイクロ秒として解釈される数値
• json - JSON オブジェクト
• binary - バイナリデータ（base64 エンコード）
• list - 値の配列
• struct - 定義済みフィールドを持つ入れ子オブジェクト

特定のストリームを確認するには、ストリーム ID またはストリーム名を指定して pipelines streams get コマンドを実行します。

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

アカウント内のすべてのストリームを一覧するには、pipelines streams list コマンドを実行します。

npx wrangler pipelines streams list

ストリーム作成後に更新できる HTTP 取り込み設定があります。ストリーム作成後のスキーマ変更はサポートしていません。

ストリームを削除するには、pipelines streams delete コマンドを実行します。

npx wrangler pipelines streams delete <STREAM_ID>