Прием данных в GeoCatalog с помощью API Bulk Ingestion

В этой статье показано, как загрузить множество геопространственных наборов данных в GeoCatalog с помощью API пакетной загрузки. Сначала мы создадим и настраиваем источник приема GeoCatalog. Создание источника загрузки устанавливает безопасное подключение между ресурсом GeoCatalog и местоположением хранилища ваших существующих геопространственных данных. Затем мы создадим коллекцию каталога доступа SpatioTemporal (STAC) в вашем ресурсе GeoCatalog для хранения данных, которые загружаются. Наконец, мы используем API пакетного ввода данных для запуска рабочего процесса загрузки. Завершив выполнение этих действий, геопространственные данные будут приняты и доступны из интерфейсов и API GeoCatalog.

Предпосылки

В подписке Azure:

Геопространственный набор данных в контейнере BLOB учётной записи хранения.

Геопространственные ресурсы данных (например, файлы GeoTIFF)
Связанные элементы STAC создают элементы STAC для этих ресурсов.
STAC-коллекция в формате JSON, ссылающаяся на все элементы STAC и геопространственные активы данных.

В локальной среде или среде разработки:

Среда Python с Python 3.10 (или более поздней версии).
Azure CLI;
Вы вошли в Azure

Microsoft Planetary Computer Pro должен иметь доступ к контейнеру хранилища BLOB-объектов Azure. В этой статье мы создаем и используем временные учетные данные токена SAS для предоставления доступа. Кроме того, эти руководства можно использовать для настройки управляемых удостоверений или жестко закодированных маркеров SAS.

REST API
Пакет SDK для Python

Создание источника приема

Создание источника приема определяет для GeoCatalog, из которого следует отправлять геопространственные данные, и какой механизм учетных данных следует использовать в рабочем процессе приема.

Установка необходимых модулей Python с помощью pip
```
pip install pystac-client azure-identity requests azure-storage-blob pyyaml
```

Импорт необходимых модулей Python

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

Задайте необходимые константы в соответствии с вашей средой

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"

Создайте маркер 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),
)

Получение токена доступа к API GeoCatalog

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

Создание полезных данных POST для исходного API приема

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

Создание источника приема путем отправки полезных данных POST в конечную точку источника приема

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

Проверка ответа

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

Замечание

Выполнение этих шагов несколько раз подряд приводит к ответу 409.

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

API приема источника не позволяет создавать несколько источников приема для одного URL-адреса контейнера. Чтобы избежать конфликтов, необходимо очистить все существующие источники приема перед созданием нового. Подробные сведения см. в разделе Очистка ресурсов.

Создание коллекции

Коллекция STAC — это высокоуровневый контейнер для элементов STAC и связанных с ними геопространственных ресурсов. В этом разделе мы создадим коллекцию STAC в нашем GeoCatalog для размещения геопространственных данных, которые мы приемываем в следующем разделе.

Импорт обязательных модулей

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

Задайте необходимые константы в соответствии с вашей средой

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

Получение токена доступа к API GeoCatalog

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

Создание базовой спецификации коллекции 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",
}
```
Замечание

Этот образец спецификации коллекции является простым примером коллекции. Дополнительные сведения о коллекциях STAC и открытом стандарте STAC см. в обзоре STAC. Дополнительные сведения о создании полной коллекции STAC см. в разделе "Создание коллекции STAC".

Создание новой коллекции с помощью API коллекции

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

Проверка результата ответа

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

Создание подключения и запуск рабочего процесса

На этом последнем шаге мы используем API приема для запуска рабочего процесса массового приема.

Импорт обязательных модулей

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

Задайте необходимые константы в соответствии с вашей средой

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

Получение токена доступа к API GeoCatalog

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

Создать тело запроса POST для API массовой загрузки

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

Отправьте пакет данных в API массового приема.

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

Проверьте ответ.

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

Проверьте состояние процесса обработки данных.

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

После завершения рабочего процесса, вы можете запрашивать, извлекать или визуализировать ваши геопространственные данные с помощью GeoCatalog STAC или API данных, либо с помощью Data Explorer. Если возникают проблемы, обратитесь к руководству по устранению неполадок или списку кодов ошибок приема.

Очистите ресурсы

Удаление источника приема

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

Установите Python SDK для Planetary Computer Pro:

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

Создание источника приема

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

Создание коллекции

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

Создание подключения и запуск рабочего процесса

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

Очистите ресурсы

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

Дополнительные сведения см. в справочнике по пакету SDK Python.

Дальнейшие шаги

Теперь, когда вы добавили несколько элементов, необходимо настроить данные для визуализации.

Конфигурация коллекции в Microsoft Planetary Computer Pro

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-06-03

Прием данных в GeoCatalog с помощью API Bulk Ingestion

Предпосылки

Создание источника приема

Создание коллекции

Создание подключения и запуск рабочего процесса

Очистите ресурсы

Дальнейшие шаги

Связанный контент

Обратная связь

Дополнительные ресурсы