Tutorial: introdução à API REST do Snowpipe Streaming usando cURL e um JWT:

Nota

Recomendamos que você comece com oSDK do Snowpipe Streaming sobre API REST para se beneficiar do melhor desempenho e da experiência de introdução.

Este guia mostra como transmitir dados para o Snowflake usando a API REST do Snowpipe Streaming e um token Web JSON (JWT) gerado com SnowSQL.

Pré-requisitos

Antes de começar, certifique-se de ter os seguintes itens:

Usuário e objetos do Snowflake:

Um usuário do Snowflake configurado para autenticação de par de chaves. Registre sua chave pública usando o seguinte comando SQL:

ALTER USER MY_USER SET RSA_PUBLIC_KEY='<your-public-key>';

Um banco de dados, um esquema e uma tabela de destino do Snowflake para ingestão de streaming. Você pode criá-los usando os seguintes comandos SQL e substituindo os espaços reservados, como MY_DATABASE, MY_SCHEMA e MY_TABLE, pelos nomes desejados:

-- Create Database and Schema
CREATE OR REPLACE DATABASE MY_DATABASE;
CREATE OR REPLACE SCHEMA MY_SCHEMA;

-- Create Target Table
CREATE OR REPLACE TABLE MY_TABLE (
    id NUMBER,
    c1 NUMBER,
    ts STRING
);

ACCOUNT_IDENTIFIER:

Sugerimos o uso do Formato 1 para o ACCOUNT_IDENTIFIER, que usa o nome da conta da sua organização; por exemplo, myorg-account123. Para obter mais informações sobre o formato, consulte Identificadores de conta.

Ferramentas instaladas:

  • curl: Para fazer solicitações HTTP.

  • jq: Para analisar respostas JSON.

  • SnowSQL: Para executar comandos, o cliente de linha de comando do Snowflake.

JWT gerado:

gere seu JWT usando SnowSQL:

snowsql --private-key-path rsa_key.p8 --generate-jwt \
  -a <ACCOUNT_IDENTIFIER> \
  -u MY_USER

Cuidado

Armazene seu JWT com segurança. Evite expô-lo em logs ou scripts.

Instruções passo a passo

Conclua as seguintes etapas para transmitir dados para o Snowflake.

Etapa 1: Definir variáveis ​ de ambiente

Configure as variáveis ​​de ambiente necessárias para sua conta Snowflake e a operação de streaming. Observe que a variável PIPE tem como alvo o canal de streaming padrão associado à sua tabela.

# Paste the JWT token obtained from SnowSQL
export JWT_TOKEN="PASTE_YOUR_JWT_TOKEN_HERE"

# Configure your Snowflake account and resources:
export ACCOUNT="<ACCOUNT_IDENTIFIER>" # For example, ab12345
export USER="MY_USER"
export DB="MY_DATABASE"
export SCHEMA="MY_SCHEMA"
export PIPE="MY_TABLE-STREAMING"
export CHANNEL="MY_CHANNEL"

# Replace ACCOUNT with your Account URL Host to form the control plane host:
export CONTROL_HOST="${ACCOUNT}.snowflakecomputing.com"

Etapa 2: Descobrir o host de ingestão

Importante

Se o nome de sua conta Snowflake contiver sublinhados (por exemplo, MY_ACCOUNT), um problema conhecido pode causar um erro interno ao chamar o serviço de ingestão.

Você deve substituir todos os sublinhados por traços no INGEST_HOST antes de gerar o token com escopo. Este formato convertido (com traços) deve ser usado para todas as chamadas de API REST subsequentes, incluindo a geração do próprio token com escopo.

Por exemplo, se o nome do host retornado for my_account.region.ingest.snowflakecomputing.com, você deve alterá-lo para my-account.region.ingest.snowflakecomputing.com em todas chamadas de API REST subsequentes.

O host de ingestão é o ponto de extremidade para o streaming de dados. Descobrir o host de ingestão usando seu JWT:

export INGEST_HOST=$(curl -sS -X GET \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -H "X-Snowflake-Authorization-Token-Type: KEYPAIR_JWT" \
  "https://${CONTROL_HOST}/v2/streaming/hostname")

echo "Ingest Host: $INGEST_HOST"

Obtenha um token com escopo para autorizar operações no host de ingestão:

export SCOPED_TOKEN=$(curl -sS -X POST "https://$CONTROL_HOST/oauth/token" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -d "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&scope=${INGEST_HOST}")

echo "Scoped Token obtained for ingest host"

Etapa 3: Abrir o canal

Abrir um canal de streaming para iniciar a ingestão de dados:

curl -sS -X PUT \
  -H "Authorization: Bearer $SCOPED_TOKEN" \
  -H "Content-Type: application/json" \
  "https://${INGEST_HOST}/v2/streaming/databases/$DB/schemas/$SCHEMA/pipes/$PIPE/channels/$CHANNEL" \
  -d '{}' | tee open_resp.json | jq .

Etapa 4: Anexar uma linha de dados

Anexar uma única linha de dados ao canal aberto.

4.1 Extrair tokens de continuação e deslocamento

Esses tokens são cruciais para manter o estado da sua sessão de streaming.

export CONT_TOKEN=$(jq -r '.next_continuation_token' open_resp.json)
export OFFSET_TOKEN=$(jq -r '.channel_status.last_committed_offset_token' open_resp.json)
export NEW_OFFSET=$((OFFSET_TOKEN + 1))

4.2 Criar linha de amostra

Gerar uma linha de dados de amostra no formato NDJSON:

export NOW_TS=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

cat <<EOF > rows.ndjson
{
  "id": 1,
  "c1": $RANDOM,
  "ts": "$NOW_TS"
}
EOF

4.3 Acrescentar linha

Enviar a linha de amostra para o canal de streaming:

curl -sS -X POST \
  -H "Authorization: Bearer $SCOPED_TOKEN" \
  -H "Content-Type: application/x-ndjson" \
  -H "Content-Encoding: zstd" \
  "https://${INGEST_HOST}/v2/streaming/data/databases/$DB/schemas/$SCHEMA/pipes/$PIPE/channels/$CHANNEL/rows?continuationToken=$CONT_TOKEN&offsetToken=$NEW_OFFSET" \
  --data-binary @rows.ndjson | jq .

Nota

Este exemplo inclui o cabeçalho Content-Encoding: zstd para demonstrar suporte à compactação. Para este exemplo simples com dados não compactados, você pode omitir esse cabeçalho. Ao enviar dados compactados, especifique zstd ou gzip para corresponder ao formato de compactação da sua carga útil.

Importante

  • Após cada operação de acréscimo, você deverá atualizar o continuationToken para a próxima chamada de acréscimo. A resposta da chamada de acréscimo de linhas contém um campo next_continuation_token que você deve usar para fazer suas atualizações.

  • O sucesso da operação de acréscimo confirma apenas que o serviço recebeu os dados, e não que eles persistem na tabela. Execute a etapa seguinte para verificar a persistência antes de consultar ou passar para o próximo lote.

4.4 Verificar a persistência dos dados e o deslocamento confirmado usando getChannelStatus

Conclua essa etapa crítica para garantir a confiabilidade do aplicativo. Não há garantia de que os dados serão persistentes até avançar o committedOffset. Para confirmar que as linhas que você acabou de anexar persistem com sucesso, use getChannelStatus.

Verifique o status atual do seu canal de streaming:

curl -sS -X POST \
  -H "Authorization: Bearer $SCOPED_TOKEN" \
  -H "Content-Type: application/json" \
  "https://${INGEST_HOST}/v2/streaming/databases/$DB/schemas/$SCHEMA/pipes/$PIPE:bulk-channel-status" \
  -d "{\"channel_names\": [\"$CHANNEL\"]}" | jq ".channel_statuses.\"$CHANNEL\""

Verificação

Você deve garantir que o committedOffset retornado na resposta seja maior ou igual ao deslocamento das linhas que acabou de anexar. Somente após o avanço de:code:committedOffset você poderá ter certeza de que os dados estão disponíveis com segurança na tabela.

4.5 Consultar se os dados persistiram na tabela

Depois de confirmar que o committedOffset avançou a etapa anterior (4.4), você poderá consultar para saber se eles foram ingeridos na tabela do Snowflake.

Execute a seguinte consulta SQL no Snowflake:

SELECT * FROM MY_DATABASE.MY_SCHEMA.MY_TABLE WHERE id = 1;

Etapa 5 (opcional): Limpeza

Remover arquivos temporários e desconfigurar variáveis ​ de ambiente:

rm -f rows.ndjson open_resp.json
unset JWT_TOKEN SCOPED_TOKEN ACCOUNT USER DB SCHEMA PIPE CHANNEL CONTROL_HOST INGEST_HOST CONT_TOKEN OFFSET_TOKEN NEW_OFFSET NOW_TS

Solução de problemas

  • HTTP 401 (Não autorizado): verifique se o seu token JWT é válido e não expirou. Se necessário, gere-o novamente.

  • HTTP 404 (Não encontrado): verifique novamente se os nomes do banco de dados, esquema, pipe e canal estão escritos corretamente e existem na sua conta Snowflake.

  • Sem host de ingestão: certifique-se de que o URL do host do plano de controle esteja correto e acessível.