Ingerir dados no GeoCatalog com a API de ingestão em massa

Este artigo mostra como ingerir muitos ativos de dados geoespaciais ao mesmo tempo em um GeoCatalog usando a API de Ingestão em Massa. Primeiro, criamos e configuramos uma Fonte de Ingestão GeoCatalog. A criação de uma fonte de ingestão estabelece uma conexão segura entre um recurso geocatalog e o local de armazenamento para seus dados geoespaciais existentes. Em seguida, criamos uma coleção STAC (SpatioTemporal Access Catalog) dentro do recurso GeoCatalog para armazenar os dados que estão sendo ingeridos. Por fim, usamos a API de Ingestão em Massa para iniciar o fluxo de trabalho de ingestão. Quando você concluir essas etapas, seus dados geoespaciais serão ingeridos e acessíveis nas interfaces e APIs do GeoCatalog.

Pré-requisitos

Em sua assinatura do Azure:

Um conjunto de dados geoespaciais no contêiner de blob da conta de armazenamento:

Ativos de dados geoespaciais (por exemplo, arquivos GeoTIFF)
Os itens STAC associados criam itens STAC para esses ativos.
Um JSON de Coleção STAC que faz referência a todos os Itens STAC e ativos de dados geoespaciais.

Em seu ambiente local/de desenvolvimento:

Um ambiente python executando o Python 3.8 ou posterior.
Azure CLI
Você está conectado ao Azure

O Computador Planetário Pro da Microsoft deve ter acesso ao contêiner do Armazenamento de Blobs do Azure. Neste artigo, criamos e usamos uma credencial temporária de token SAS para conceder esse acesso. Como alternativa, você pode usar esses guias para configurar a Identidade Gerenciada ou tokens SAS codificados.

Criar fonte de ingestão

Criar uma fonte de ingestão determina para o GeoCatalog qual fonte será usada para ingestão de dados geoespaciais e qual mecanismo de credencial será utilizado no fluxo de trabalho de ingestão.

Instalar os módulos do Python necessários usando pip

pip install pystac-client azure-identity requests azure-storage-blob pyyaml

Importar os módulos do Python necessários

import os
import requests
from azure.identity import AzureCliCredential
from datetime import datetime, timedelta, timezone
import azure.storage.blob
from urllib.parse import urlparse
import yaml

Definir as constantes necessárias de acordo com seu ambiente

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
CONTAINER_URI = "<container_uri>" # The URI for the blob storage container housing your geospatial data
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2025-04-30-preview"

Criar um token SAS

# Parse the container URL
parsed_url = urlparse(CONTAINER_URI)
account_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
account_name = parsed_url.netloc.split(".")[0]
container_name = parsed_url.path.lstrip("/")

credential = azure.identity.AzureCliCredential()
blob_service_client = azure.storage.blob.BlobServiceClient(
    account_url=account_url,
    credential=credential,
)

now = datetime.now(timezone.utc).replace(microsecond=0)
key = blob_service_client.get_user_delegation_key(
    key_start_time=now + timedelta(hours=-1),
    key_expiry_time=now + timedelta(hours=1),
)

sas_token = azure.storage.blob.generate_container_sas(
    account_name=account_name,
    container_name=container_name,
    user_delegation_key=key,
    permission=azure.storage.blob.ContainerSasPermissions(
        read=True,
        list=True,
    ),
    start=now + timedelta(hours=-1),
    expiry=now + timedelta(hours=1),
)

Obter o token de acesso à API do GeoCatalog

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Criar conteúdo POST para a API de origem de ingestão

# Payload for the POST request
payload = {
    "Kind": "SasToken",
    "connectionInfo": {
        "containerUrl": CONTAINER_URI,
        "sasToken": sas_token,
    },
}

Criar a fonte de ingestão enviando o conteúdo POST para o ponto de extremidade de origem de ingestão

# STAC Collection API endpoint
endpoint = f"{GEOCATALOG_URI}/inma/ingestion-sources"

# Make the POST request
response = requests.post(
    endpoint,
    json=payload,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

Verificar a resposta

# Print the response
if response.status_code == 201:
    print("Ingestion source created successfully")
    ingestion_source_id = response.json().get("id") #saved for later to enable resoource clean up
else:
    print(f"Failed to create ingestion: {response.text}")

Observação

Executar estas etapas mais de uma vez resulta em uma resposta 409:

Container url <container uri> already contains a SAS token ingestion source with id <sas token id>

A API de origem de ingestão não permite que você crie mais de uma fonte de ingestão para a mesma URL de contêiner. Para evitar conflitos, certifique-se de limpar todas as fontes de ingestão existentes antes de criar uma nova. Para obter detalhes, confira Limpar recursos.

Criar coleção

Uma coleção STAC é o contêiner de alto nível para itens STAC e seus ativos geoespaciais associados. Nesta seção, estamos criando uma coleção STAC em nosso GeoCatalog para abrigar os dados geoespaciais que estamos ingerindo na próxima seção.

Importe os módulos necessários

import os
import requests
import yaml
from pprint import pprint
from azure.identity import AzureCliCredential

Definir as constantes necessárias de acordo com seu ambiente

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2025-04-30-preview"

COLLECTION_ID = "example-collection" #You can your own collection ID
COLLECTION_TITLE = "Example Collection" #You can your own collection title

Obter o token de acesso à API do GeoCatalog

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Criar uma especificação básica da coleção STAC

collection = {
    "id": COLLECTION_ID,
    "type": "Collection",
    "title": COLLECTION_TITLE,
    "description": "An example collection",
    "license": "CC-BY-4.0",
    "extent": {
        "spatial": {"bbox": [[-180, -90, 180, 90]]},
        "temporal": {"interval": [["2018-01-01T00:00:00Z", "2018-12-31T23:59:59Z"]]},
    },
    "links": [],
    "stac_version": "1.0.0",
    "msft:short_description": "An example collection",
}

Observação

Esta especificação de amostra é um exemplo básico de uma coleção. Para obter mais informações sobre coleções STAC e o padrão aberto STAC, consulte a visão geral do STAC. Para obter mais informações sobre como criar uma coleção STAC completa, consulte Create STAC Collection.

Criar a nova coleção com a API de Coleção

response = requests.post(
    f"{GEOCATALOG_URI}/stac/collections",
    json=collection,
    headers={"Authorization": "Bearer " + access_token.token},
    params={"api-version": API_VERSION},
)

Verificar o resultado da resposta

if response.status_code == 201:
    print("Collection created successfully")
    pprint(response.json())
else:
    print(f"Failed to create ingestion: {response.text}")

Criar conexão e executar fluxo de trabalho

Nesta etapa final, estamos usando a API de ingestão para iniciar um fluxo de trabalho de ingestão em massa.

Importar módulos necessários

import os
import requests
import yaml
from azure.identity import AzureCliCredential

Definir as constantes necessárias de acordo com seu ambiente

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2025-04-30-preview"

COLLECTION_ID = "example-collection" #You can your own collection ID
catalog_href = "<catalog_href>" #The blob storage location of the STAC Catalog JSON file

skip_existing_items = False
keep_original_assets = False
timeout_seconds = 300

Obter o token de acesso à API do GeoCatalog

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Criar o conteúdo POST para a API de Ingestão em Massa

url = f"{GEOCATALOG_URI}/inma/collections/{COLLECTION_ID}/ingestions"
body = {
    "importType": "StaticCatalog",
    "sourceCatalogUrl": catalog_href,
    "skipExistingItems": skip_existing_items,
    "keepOriginalAssets": keep_original_assets,
}

Envie o conteúdo para a API de Ingestão em Massa.

ing_response = requests.post(
    url,
    json=body,
    timeout=timeout_seconds,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

Verifique a resposta.

if ing_response.status_code == 201:
    print("Ingestion created successfully")
    ingestion_id = ing_response.json()["ingestionId"]
    print(f"Created ingestion with ID: {ingestion_id}")
else:
    print(f"Failed to create ingestion: {ing_response.text}")

Verifique o status do fluxo de trabalho de ingestão.

runs_endpoint = (
    f"{geocatalog_url}/inma/collections/{collection_id}/ingestions/{ingestion_id}/runs"
)

wf_response = requests.post(
    runs_endpoint,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

if wf_response.status_code == 201:
    print("Workflow started successfully")
else:
    print(f"Failed to create ingestion run: {wf_response.text}")

Depois que o fluxo de trabalho for concluído, você poderá consultar, recuperar ou visualizar seus dados geoespaciais usando o GEOCatalog STAC ou APIs de dados ou com o Data Explorer.

Limpar os recursos

Excluir fonte de ingestão

del_is_endpoint = f"{GEOCATALOG_URI}/inma/ingestion-sources/{ingestion_source_id}"
del_is_response = requests.delete(
    del_is_endpoint,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

if del_is_response.status_code == 200:
    print("Ingestion source deleted successfully")
else:
    print(f"Failed to delete ingestion source")

Próximas etapas

Agora que você adicionou alguns itens, você deve configurar os dados para visualização.

Configuração da coleção no Microsoft Planetary Computer Pro

Comentários

Esta página foi útil?

Last updated on 2025-05-20

Compartilhar via

Ingerir dados no GeoCatalog com a API de ingestão em massa

Pré-requisitos

Criar fonte de ingestão

Criar coleção

Criar conexão e executar fluxo de trabalho

Limpar os recursos

Próximas etapas

Conteúdo relacionado

Comentários

Recursos adicionais