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

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

Создание первой Бакет R2 и включив её каталог данных.
Создание 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. Создайте bucket R2

Если вы ещё не выполнили вход, выполните команду:
```
npx wrangler login
```

Создайте бакет R2:

npx wrangler r2 bucket create pipelines-tutorial

На панели управления Cloudflare перейдите к разделу Объектное хранилище R2 страницу.
Перейдите в Обзор ↗
Выберите Создание бакета.
Введите имя бакета: pipelines-tutorial
Выберите Создание бакета.

2. Включите R2 Data Catalog

Включите каталог для своего бакета R2:

npx wrangler r2 bucket catalog enable pipelines-tutorial

При выполнении этой команды запишите значения "Warehouse" и "Catalog URI". Они понадобятся вам позже.

На панели управления Cloudflare перейдите к разделу Объектное хранилище R2 страницу.
Перейдите в Обзор ↗
Выберите бакет: pipelines-tutorial.
Перейдите на вкладку Настройки вкладку и прокрутите вниз до R2 Data Catalog, и выберите Включить.
После включения обратите внимание на Catalog URI и Имя хранилища.

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

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

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

4. Создайте 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 Table
- Имя бакета R2: pipelines-tutorial
- Пространство имён: default
- Имя таблицы: ecommerce
- Catalog API token: введите токен, полученный на шаге 3
- Сжатие: zstd
- Переключать файл при достижении размера (МБ): 100
- Переключать файл по истечении времени (с): 10 (для более быстрого отображения данных в этом руководстве)
SQL-преобразование: Выберите Use simple ingestion query для использования:
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream
```

После завершения настройки запишите URL-адрес конечной точки HTTP, указанный в итоговом выводе.

В панели управления Cloudflare перейдите в 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": "f64",
			"required": false
		}
	]
}

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

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

5. Отправьте тестовые данные

Отправка событий электронной коммерции на 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.

6. Проверьте данные в своём bucket

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

7. Запросите свои данные с помощью 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 это токен, который вы создали в шаге 3. Дополнительные сведения об установке переменных окружения см. в Системные переменные окружения 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 из шага 2.

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