Ingesta de datos en GeoCatalog con la API de Ingesta en Masa

En este artículo se muestra cómo ingerir muchos recursos de datos geoespaciales a la vez en un geocatalog mediante la API de ingesta masiva. En primer lugar, creamos y configuramos un origen de ingesta de GeoCatalog. La creación de un origen de ingesta establece una conexión segura entre un recurso GeoCatalog y la ubicación de almacenamiento de los datos geoespaciales existentes. A continuación, creamos una colección de catálogos de acceso spatiotemporales (STAC) dentro del recurso GeoCatalog para almacenar los datos que se ingieren. Por último, usamos la API de ingestión masiva para iniciar el flujo de trabajo de ingestión. Cuando haya terminado de completar estos pasos, los datos geoespaciales se integran y son accesibles desde las interfaces de usuario y las API de GeoCatalog.

Prerrequisitos

En la suscripción de Azure:

Un conjunto de datos geoespacial en el contenedor de blobs de la cuenta de almacenamiento:

Recursos de datos geoespaciales (por ejemplo, archivos GeoTIFF)
Los elementos de STAC asociados crean elementos de STAC para estos recursos.
JSON de la colección STAC que hace referencia a todos los recursos de datos geoespaciales y elementos de STAC.

En el entorno local o de desarrollo:

Un entorno de Python que ejecuta Python 3.10 (o posterior).
CLI de Azure
Ha iniciado sesión en Azure

Ordenador Planetario de Microsoft Pro debe tener acceso al contenedor de Azure Blob Storage. En este artículo, creamos y usamos una credencial de token de SAS temporal para conceder este acceso. Como alternativa, puede usar estas guías para configurar identidad administrada o tokens saS codificados de forma rígida.

REST API
SDK de Python

Crear origen de ingesta

La creación de un origen de ingesta define para GeoCatalog el origen del que ingerir datos geoespaciales y el mecanismo de credenciales que se va a usar en el flujo de trabajo de ingesta.

Instalación de los módulos de Python necesarios mediante pip

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

Importación de los módulos de Python necesarios

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

Establecer las constantes necesarias según el entorno

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 = "2026-04-15"

Creación del token de 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),
)

Obtención del token de acceso de geocatalog API

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

Creación de una carga POST para la API de origen de ingesta

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

Cree la fuente de ingesta mediante el envío de la carga POST al punto de conexión de la fuente de ingesta.

# 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},
)

Comprobación de la respuesta

# 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 resource clean up
else:
    print(f"Failed to create ingestion: {response.text}")

Nota:

Al ejecutar estos pasos más de una vez, se devuelve una respuesta 409:

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

La API de origen de ingesta no permite crear más de un origen de ingesta para la misma dirección URL del contenedor. Para evitar conflictos, asegúrese de limpiar los orígenes de ingesta existentes antes de crear uno nuevo. Para obtener más información, vea Limpieza de recursos.

Crear una colección

Una colección STAC es el contenedor de alto nivel para los elementos de STAC y sus recursos geoespaciales asociados. En esta sección, vamos a crear una colección de STAC dentro de nuestro GeoCatalog para hospedar los datos geoespaciales que estamos ingeriendo en la sección siguiente.

Importar los módulos necesarios

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

Establecer las constantes necesarias según el entorno

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 = "2026-04-15"

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

Obtención del token de acceso de geocatalog API

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

Creación de una especificación básica de colección 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",
}

Nota:

Esta especificación de colección de muestra es un ejemplo básico de una colección. Para obtener más información sobre las colecciones STAC y el estándar abierto STAC, consulte la información general de STAC. Para obtener más información sobre cómo crear una colección completa de STAC, consulte Creación de una colección de STAC.

Creación de la nueva colección con collection API

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

Comprobación del resultado de la respuesta

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

Crear conexión y ejecutar flujo de trabajo

En este paso final, se usa la API de ingesta para iniciar un flujo de trabajo de ingesta masiva.

Importación de módulos necesarios

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

Establecer las constantes necesarias según el entorno

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 = "2026-04-15"

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

Obtención del token de acceso de geocatalog API

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

Creación de la carga POST para la API de Ingesta masiva

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

Envíe la carga a la API de Ingesta masiva.

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

Compruebe la respuesta.

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}")

Compruebe el estado del flujo de trabajo de ingesta.

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}")

Una vez completado el flujo de trabajo, puede consultar, recuperar o visualizar los datos geoespaciales mediante geocatalog STAC o las API de datos, o con el Explorador de datos. Si encuentra problemas, consulte la guía de solución de problemas o la lista de códigos de error de ingesta.

Limpieza de recursos

Eliminar origen de ingesta

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")

Instale el SDK de Python de Planetary Computer Pro:

pip install azure-planetarycomputer azure-identity azure-storage-blob

Crear origen de ingesta

from azure.planetarycomputer import PlanetaryComputerProClient
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse
import azure.storage.blob

client = PlanetaryComputerProClient(
    endpoint="<your-geocatalog-url>",
    credential=DefaultAzureCredential()
)

CONTAINER_URI = "<container_uri>"

# Generate SAS token using azure-storage-blob (same as REST API approach)
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 = DefaultAzureCredential()
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),
)

# Create the ingestion source
ingestion_source = client.ingestion.create_source({
    "kind": "SasToken",
    "connectionInfo": {
        "containerUrl": CONTAINER_URI,
        "sasToken": sas_token,
    },
})
print(f"Ingestion source created: {ingestion_source['id']}")

Crear una colección

collection = {
    "id": "example-collection",
    "type": "Collection",
    "title": "Example Collection",
    "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",
}

poller = client.stac.begin_create_collection(body=collection)
result = poller.result()
print(f"Collection created: {result}")

Crear conexión y ejecutar flujo de trabajo

COLLECTION_ID = "example-collection"
catalog_href = "<catalog_href>"  # Blob storage location of the STAC Catalog JSON file

ingestion = client.ingestion.create(COLLECTION_ID, {
    "importType": "StaticCatalog",
    "sourceCatalogUrl": catalog_href,
    "skipExistingItems": False,
    "keepOriginalAssets": False,
})
ingestion_id = ingestion.id
print(f"Created ingestion with ID: {ingestion_id}")

# Start the ingestion workflow
run = client.ingestion.create_run(COLLECTION_ID, ingestion_id)
print("Workflow started successfully")

Limpieza de recursos

client.ingestion.delete_source(ingestion_source["id"])
print("Ingestion source deleted successfully")

Para obtener más información, consulte la referencia del SDK de Python.

Pasos siguientes

Ahora que ha agregado algunos elementos, debe configurar los datos para su visualización.

Configuración de la colección en Ordenador Planetario de Microsoft Pro

Comentarios

¿Le resultó útil esta página?

Última actualización el 2026-06-03

Ingesta de datos en GeoCatalog con la API de Ingesta en Masa

Prerrequisitos

Crear origen de ingesta

Crear una colección

Crear conexión y ejecutar flujo de trabajo

Limpieza de recursos

Pasos siguientes

Contenido relacionado

Comentarios

Recursos adicionales