← Cloudflare Pipelines / pipelines / streams

Управление потоками

Вы узнаете, как:

Создание и настройка streams для приёма данных
Просмотр и изменение настроек потока
Удаление streams, которые больше не нужны

Создайте stream

Потоки становятся доступны конвейерам в виде SQL таблиц по имени потока (например, SELECT * FROM my_stream).

Dashboard

На панели управления Cloudflare перейдите к разделу Pipelines страницу.
Перейдите в Pipelines ↗
Выберите Create Pipeline чтобы запустить мастер создания pipeline.
Завершите работу с мастером, чтобы создать stream вместе со связанными sink и pipeline.

Wrangler CLI

Чтобы создать поток, выполните pipelines streams create команда:

npx wrangler pipelines streams create <STREAM_NAME>

Как вариант можно воспользоваться интерактивным мастером настройки, который поможет настроить stream, sink и pipeline. Для этого выполните pipelines setup команда:

npx wrangler pipelines setup

Настройка схемы

Потоки поддерживают два подхода к обработке данных:

Структурированные потоки: Определите схему с конкретными полями и типами данных. События проверяются на соответствие этой схеме.
Неструктурированные потоки: Принимает любой JSON без валидации. Такие streams имеют одно value столбец, содержащий данные JSON.

Чтобы создать структурированный поток, укажите файл схемы:

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 - вложенные объекты с заданными полями

Просмотр конфигурации потока

Dashboard

В панели управления Cloudflare перейдите в Pipelines > Streams.
Выберите поток, чтобы просмотреть его конфигурацию.

Wrangler CLI

Чтобы просмотреть конкретный поток, выполните pipelines streams get команду, указав либо ID потока, либо его имя:

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

Чтобы получить список всех потоков в вашем аккаунте, выполните pipelines streams list команда:

npx wrangler pipelines streams list

Обновление настроек приёма HTTP

После создания потока вы можете изменить некоторые настройки приёма данных по HTTP. Изменение схемы после создания потока не поддерживается.

Dashboard

В панели управления Cloudflare перейдите в Pipelines > Streams.
Выберите поток, который нужно обновить.
В Настройки вкладке перейдите в HTTP Ingest.
Чтобы включить или выключить приём данных по HTTP, выберите Включить или Отключить.
Чтобы изменить настройки аутентификации и CORS, выберите Изменить и изменить.
Сохраните свои изменения.

Удалить поток

Dashboard

В панели управления Cloudflare перейдите в Pipelines > Streams.
Выберите поток, который нужно удалить.
В Настройки вкладке перейдите в Общее, и выберите Удалить.

Wrangler CLI

Чтобы удалить поток, выполните pipelines streams delete команда:

npx wrangler pipelines streams delete <STREAM_ID>