Začínáme

Tento návod vás provede:

Vytvoření API token potřebná k tomu, aby se Pipelines mohly ověřit vůči vašemu datovému katalogu.
Vytvoření prvního pipelinu s jednoduchým schématem e-shopu, který zapisuje do Apache Iceberg ↗ tabulku spravovanou službou R2 Data Catalog.
Odeslání ukázkových e-commerce dat přes HTTP endpoint.
Ověřování dat ve vašem bucketu a jejich dotazování pomocí R2 SQL.

Předpoklady

Zaregistrujte si účet Cloudflare ↗.
Nainstalujte Node.js ↗.

Správce verzí Node.js

Použijte správce verzí Node, jako je Volta ↗ nebo nvm ↗ abyste se vyhnuli problémům s oprávněními a mohli měnit verze Node.js. Wrangler, o kterém se dozvíte dále v této příručce, vyžaduje verzi Node 16.17.0 nebo novější.

1. Vytvořte token API

Pipelines se musí k R2 Data Catalog ověřit pomocí Token R2 API s oprávněními pro katalog a R2.

V dashboardu Cloudflare přejděte na Objektové úložiště R2 stránce.
Přejděte na Přehled ↗
Vyberte Správa API tokenů.
Vyberte Vytvoření API tokenu účtu.
Zadejte název pro váš API token.
V části Oprávnění, vyberte Admin Read & Write oprávnění.
Vyberte Create Account API Token.
Všimněte si Hodnota tokenu.

2. Vytvořte svou první pipeline

Nejprve vytvořte soubor schématu, který definuje strukturu vašich e-commerce dat:

Vytvořit 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
		}
	]
}

Pomocí interaktivního nastavení vytvořte pipeline, která zapisuje do R2 Data Catalog:

npx wrangler pipelines setup

Postupujte podle výzev:

Název pipeline: Zadejte ecommerce
Konfigurace streamu:
- Povolit HTTP endpoint: yes
- Vyžadovat ověření: no (pro jednoduchost)
- Nakonfigurujte vlastní CORS origins: no
- Definice schématu: Load from file
- Cesta k souboru schématu: schema.json (nebo cesta k vašemu souboru)
Konfigurace sinku:
- Typ cíle: Data Catalog (Iceberg)
- Režim nastavení: Simple (recommended defaults)
- Název R2 bucketu: pipelines-tutorial (vytvoří se automaticky, pokud neexistuje)
- Název tabulky: ecommerce
- Catalog API token: Zadejte token z kroku 1
Zkontrolovat: Potvrďte souhrn a vyberte Create resources
Transformace SQL: Vyberte Simple ingestion (SELECT * FROM stream)

Možnosti pokročilého režimu

Pokud vyberete Pokročilé místo Jednoduché si během konfigurace sinku můžete přizpůsobit následující doplňkové možnosti:

Formát: Formát výstupního souboru (například Parquet)
Komprese: Kompresní algoritmus (například zstd)
Zásady rotace: Práh velikosti souboru (minimálně 5 MB) a časový interval (minimálně 10 sekund) pro vytváření nových souborů
Přihlašovací údaje: Zvolte mezi automatickým generováním přihlašovacích údajů nebo jejich ručním zadáním pro R2
Namespace: Obor názvů Data Catalog (výchozí hodnota default)

Po dokončení nastavení příkaz vypíše konfigurační úryvek pro soubor Wrangler, příklad Worker bindingu se vzorovými daty a příkaz curl pro HTTP endpoint. Poznamenejte si URL adresu HTTP endpointu a pipelines konfiguraci pro použití v následujících krocích.

Název pipeline můžete také předem nastavit pomocí --name příznak:

npx wrangler pipelines setup --name ecommerce

V Cloudflare dashboardu přejděte na Objektové úložiště R2.
Přejděte na Přehled ↗
Vyberte Vytvořit bucket a zadejte název bucketu: pipelines-tutorial.
Vyberte Vytvořit bucket.
Vyberte bucket, přepněte na kartu Nastavení kartu a přejděte dolů na R2 Data Catalog, a vyberte Povolit.
Po zapnutí si poznamenejte Catalog URI a Název warehouse.
Přejděte na Pipelines > Pipelines.
Přejděte na Pipelines ↗
Vyberte Create Pipeline.
Připojení ke streamu:
- Název pipeline: ecommerce
- Povolit HTTP endpoint pro odesílání dat: Povoleno
- HTTP ověřování: Disabled (výchozí)
- Vyberte Next

Define Input Schema:

Vyberte JSON editor

Zkopírujte schéma:

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

Vyberte Next

Define Sink:
- Vyberte svůj bucket R2: pipelines-tutorial
- Typ úložiště: R2 Data Catalog
- Namespace: default
- Název tabulky: ecommerce
- Pokročilá nastavení: Změna Maximum Time Interval na 10 seconds
- Vyberte Next
Přihlašovací údaje:
- Zakázat Automaticky vytvořit API token účtu pro váš sink
- Zadejte Catalog Token z kroku 1
- Vyberte Next
Pipeline Definition:
- Ponechte výchozí SQL dotaz:
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream;
```
- Vyberte Create Pipeline
Po vytvoření pipeline si poznamenejte ID streamu pro další krok.

3. Odešlete ukázková data

Odešlete e-commerce události do HTTP endpointu vaší pipeline:

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
    }
  ]'

Nahraďte {stream-id} se skutečným koncovým bodem streamu z nastavení pipeline.

4. Ověřte data ve vašem bucketu

V dashboardu Cloudflare přejděte na Objektové úložiště R2 stránce.
Vyberte svůj bucket: pipelines-tutorial.
Měli byste vidět soubory s metadaty Iceberg a datové soubory vytvořené vaší pipeline. Pokud ve svém bucketu žádné soubory nevidíte, počkejte několik minut a zkuste to znovu.
Data jsou uspořádána ve formátu Apache Iceberg, přičemž metadata sledují verze tabulky.

5. Dotazujte svá data pomocí R2 SQL

Nastavte prostředí pro používání R2 SQL:

export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

Nebo vytvořte .env soubor pomocí:

WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

Kde YOUR_API_TOKEN je token, který jste vytvořili v kroku 1. Další informace o nastavení proměnných prostředí najdete v Systémové proměnné prostředí Wrangler.

Dotaz na vaše data:

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"

Nahraďte YOUR_WAREHOUSE_NAME s názvem warehouse, který jste si poznamenali při nastavení pipeline. Najdete jej v Cloudflare dashboardu v Objektové úložiště R2 > váš bucket > Nastavení > R2 Data Catalog.

Tuto tabulku můžete dotazovat i pomocí libovolného enginu, který podporuje Apache Iceberg. Další informace o připojení dalších enginů k R2 Data Catalog najdete v Připojení k enginům Iceberg.

Začínáme

Předpoklady

1. Vytvořte token API

2. Vytvořte svou první pipeline

3. Odešlete ukázková data

4. Ověřte data ve vašem bucketu

5. Dotazujte svá data pomocí R2 SQL

Zjistit více

Streams

Pipelines

Sinks

Příklady