Manage streams

Learn how to:

• Create and configure streams for data ingestion
• View and update stream settings
• Delete streams when no longer needed

Streams are made available to pipelines as SQL tables using the stream name (for example, SELECT * FROM my_stream).

To create a stream, run the pipelines streams create command:

npx wrangler pipelines streams create <STREAM_NAME>

Alternatively, to use the interactive setup wizard that helps you configure a stream, sink, and pipeline, run the pipelines setup command:

npx wrangler pipelines setup

Streams support two approaches for handling data:

• Structured streams: Define a schema with specific fields and data types. Events are validated against the schema.
• Unstructured streams: Accept any valid JSON without validation. These streams have a single value column containing the JSON data.

To create a structured stream, provide a schema file:

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

Example schema file:

{
	"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
				}
			]
		}
	]
}

Supported data types:

• string - Text values
• int32, int64 - Integer numbers
• float32, float64 - Floating-point numbers
• bool - Boolean true/false
• timestamp - RFC 3339 timestamps, or numeric values parsed as Unix seconds, milliseconds, or microseconds (depending on unit)
• json - JSON objects
• binary - Binary data (base64-encoded)
• list - Arrays of values
• struct - Nested objects with defined fields

To view a specific stream, run the pipelines streams get command with either the stream ID or stream name:

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

To list all streams in your account, run the pipelines streams list command:

npx wrangler pipelines streams list

You can update certain HTTP ingest settings after stream creation. Schema modifications are not supported once a stream is created.

To delete a stream, run the pipelines streams delete command:

npx wrangler pipelines streams delete <STREAM_ID>