Начало работы

Это руководство проведёт вас через следующие шаги:

Создание API-токен необходимые pipelines для аутентификации в вашем каталоге данных.
Создание первого pipeline с простой схемой интернет-магазина, которая записывает данные в Apache Iceberg ↗ таблицу, управляемую R2 Data Catalog.
Отправка примера данных электронной коммерции через HTTP-эндпоинт.
Проверка данных в вашем бакете и их запрос с помощью R2 SQL.

Предварительные требования

Зарегистрируйтесь для получения Аккаунт Cloudflare ↗.
Установка Node.js ↗.

менеджер версий Node.js

Используйте менеджер версий Node, например Volta ↗ или nvm ↗ чтобы избежать проблем с правами доступа и переключать версии Node.js. Wrangler, о котором пойдёт речь далее в этом руководстве, требует версию Node 16.17.0 или более поздней версии.

1. Создайте токен API

Pipelines должен аутентифицироваться в R2 Data Catalog с помощью R2 API-токен у которого есть разрешения на использование каталога и R2.

На панели управления Cloudflare перейдите к разделу Объектное хранилище R2 страницу.
Перейдите в Обзор ↗
Выберите Управление API-токенами.
Выберите Создание Account API token.
Присвойте токену API имя.
В разделе Разрешения, выберите Admin Read & Write разрешение.
Выберите Create Account API Token.
Обратите внимание на Значение токена.

2. Создание первого pipeline

Сначала создайте файл схемы, описывающий структуру данных вашего интернет-магазина:

Создайте schema.json:

{
	"fields": [
		{
			"name": "user_id",
			"type": "string",
			"required": true
		},
		{
			"name": "event_type",
			"type": "string",
			"required": true
		},
		{
			"name": "product_id",
			"type": "string",
			"required": false
		},
		{
			"name": "amount",
			"type": "float64",
			"required": false
		}
	]
}

Используйте интерактивную настройку, чтобы создать конвейер, который записывает данные в R2 Data Catalog:

npx wrangler pipelines setup

Следуйте подсказкам:

Имя конвейера: Введите ecommerce
Конфигурация потока:
- Включить HTTP-эндпоинт: yes
- Требуется аутентификация: no (для простоты)
- Настройте пользовательские источники CORS: no
- Определение схемы: Load from file
- Путь к файлу схемы: schema.json (или путь к вашему файлу)
Настройка приёмника:
- Тип назначения: Data Catalog (Iceberg)
- Режим настройки: Simple (recommended defaults)
- Имя бакета R2: pipelines-tutorial (создается автоматически, если не существует)
- Имя таблицы: ecommerce
- Catalog API token: введите токен, полученный на шаге 1
Проверить: Подтвердите сводку и выберите Create resources
SQL-преобразование: Выберите Simple ingestion (SELECT * FROM stream)

Параметры расширенного режима

Если вы выберете Продвинутый уровень вместо Простой при настройке приёмника вы можете задать следующие дополнительные параметры:

Формат: Формат выходного файла (например, Parquet)
Сжатие: Алгоритм сжатия (например, zstd)
Политика ротации: Пороговое значение размера файла (не менее 5 MB) и временной интервал (не менее 10 секунд) для создания новых файлов
Учётные данные: Выберите автоматическую генерацию учётных данных или ручной ввод учётных данных R2
Пространство имён: Пространство имён Data Catalog (по умолчанию default)

После завершения настройки команда выводит фрагмент конфигурации для файла Wrangler, пример Worker binding с примерами данных и команду curl для HTTP-эндпоинта. Запишите URL HTTP-эндпоинта и pipelines конфигурацию для использования в следующих шагах.

Вы также можете заранее задать имя конвейера с помощью --name флаг:

npx wrangler pipelines setup --name ecommerce

В панели управления Cloudflare перейдите в Объектное хранилище R2.
Перейдите в Обзор ↗
Выберите Создание бакета и введите имя бакета: pipelines-tutorial.
Выберите Создание бакета.
Выберите бакет, перейдите на вкладку Настройки вкладку и прокрутите вниз до R2 Data Catalog, и выберите Включить.
После включения обратите внимание на Catalog URI и Имя хранилища.
Перейдите в Pipelines > Pipelines.
Перейдите в Pipelines ↗
Выберите Create Pipeline.
Подключение к Stream:
- Имя конвейера: ecommerce
- Включить HTTP-эндпоинт для отправки данных: Enabled
- HTTP authentication: Disabled (default)
- Выберите Далее

Define Input Schema:

Выберите Редактор JSON

Скопируйте схему:

{
	"fields": [
		{
			"name": "user_id",
			"type": "string",
			"required": true
		},
		{
			"name": "event_type",
			"type": "string",
			"required": true
		},
		{
			"name": "product_id",
			"type": "string",
			"required": false
		},
		{
			"name": "amount",
			"type": "float64",
			"required": false
		}
	]
}

Выберите Далее

Define Sink:
- Выберите свой бакет R2: pipelines-tutorial
- Тип хранилища: R2 Data Catalog
- Пространство имён: default
- Имя таблицы: ecommerce
- Дополнительные настройки: Изменение Максимальный интервал времени к 10 seconds
- Выберите Далее
Учётные данные:
- Отключить Автоматически создать Account API token для вашего sink
- Введите Catalog Token из шага 1
- Выберите Далее
Определение конвейера:
- Оставьте SQL-запрос по умолчанию:
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream;
```
- Выберите Create Pipeline
После создания pipeline запомните ID потока для следующего шага.

3. Отправка тестовых данных

Отправка событий электронной коммерции на HTTP-эндпоинт вашего конвейера:

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[
    {
      "user_id": "user_12345",
      "event_type": "purchase",
      "product_id": "widget-001",
      "amount": 29.99
    },
    {
      "user_id": "user_67890",
      "event_type": "view_product",
      "product_id": "widget-002"
    },
    {
      "user_id": "user_12345",
      "event_type": "add_to_cart",
      "product_id": "widget-003",
      "amount": 15.50
    }
  ]'

Замените {stream-id} с реальной конечной точкой stream из настройки pipeline.

4. Проверка данных в bucket

На панели управления Cloudflare перейдите к разделу Объектное хранилище R2 страницу.
Выберите свой бакет: pipelines-tutorial.
Вы должны увидеть файлы метаданных и данных Iceberg, созданные вашим конвейером. Если в бакете нет файлов, подождите пару минут и повторите попытку.
Данные организованы в формате Apache Iceberg, метаданные которого отслеживают версии таблицы.

5. Запрос данных с помощью R2 SQL

Настройте окружение для работы с R2 SQL:

export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

Или создайте .env файл с помощью:

WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

Где YOUR_API_TOKEN это токен, который вы создали на шаге 1. Подробнее о настройке переменных окружения см. в Системные переменные окружения Wrangler.

Запросите свои данные:

npx wrangler r2 sql query "YOUR_WAREHOUSE_NAME" "
SELECT
    user_id,
    event_type,
    product_id,
    amount
FROM default.ecommerce
WHERE event_type = 'purchase'
LIMIT 10"

Замените YOUR_WAREHOUSE_NAME с именем warehouse, указанным при настройке pipeline. Его можно найти на панели управления Cloudflare в разделе Объектное хранилище R2 > ваш бакет > Настройки > R2 Data Catalog.

Вы также можете запрашивать эту таблицу с помощью любого движка, поддерживающего Apache Iceberg. Подробнее о подключении других движков к R2 Data Catalog см. в Подключение к движкам Iceberg.

Начало работы

Предварительные требования

1. Создайте токен API

2. Создание первого pipeline

3. Отправка тестовых данных

4. Проверка данных в bucket

5. Запрос данных с помощью R2 SQL

Подробнее

Streams

Pipelines

Приёмники

Примеры