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 basin pipelines streams create command:

npx wrangler basin pipelines streams create <STREAM_NAME>

yarn wrangler basin pipelines streams create <STREAM_NAME>

pnpm wrangler basin pipelines streams create <STREAM_NAME>

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

npx wrangler basin pipelines setup

yarn wrangler basin pipelines setup

pnpm wrangler basin 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 basin pipelines streams create my-stream --schema-file schema.json

yarn wrangler basin pipelines streams create my-stream --schema-file schema.json

pnpm wrangler basin 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 basin pipelines streams get command with either the stream ID or stream name:

npx wrangler basin pipelines streams get <STREAM_NAME_OR_ID>

yarn wrangler basin pipelines streams get <STREAM_NAME_OR_ID>

pnpm wrangler basin pipelines streams get <STREAM_NAME_OR_ID>

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

npx wrangler basin pipelines streams list

yarn wrangler basin pipelines streams list

pnpm wrangler basin 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 basin pipelines streams delete command:

npx wrangler basin pipelines streams delete <STREAM_ID>

yarn wrangler basin pipelines streams delete <STREAM_ID>

pnpm wrangler basin pipelines streams delete <STREAM_ID>