¿Qué hace dm-limits-and-best-practices?
Habilidad de referencia para las mejores prácticas de la API de Data Modeling de CDF. Cubre límites de concurrencia (evitando errores 429), patrones de paginación para instances.list e instances.query, operaciones de escritura por lotes, guía de búsqueda vs filtro y la utilidad QueuedTaskRunner (Semáforo) para controlar solicitudes concurrentes. Desencadenantes: Límites de DMS, error 429, límite de velocidad, paginación, cursor, nextCursor, procesamiento por lotes, semáforo, QueuedTaskRunner, cdfTaskRunner, instances.search, instances.list, instances.query, instances.upsert, concurrencia, bloqueo.
Introducción
Al trabajar con la API de Data Modeling (DMS) de CDF, es importante seguir las mejores prácticas para garantizar un rendimiento y una fiabilidad óptimos. Esta habilidad proporciona orientación sobre cómo manejar la concurrencia, la paginación, el procesamiento por lotes y otras operaciones comunes.
Límites de concurrencia y errores 429
La API de DMS tiene límites de concurrencia para proteger el sistema de ser sobrecargado. Cuando se excede un límite, la API devuelve un error HTTP 429 (Demasiadas solicitudes). Para evitar esto:
- Utiliza la utilidad
QueuedTaskRunner(Semáforo) para controlar el número de solicitudes concurrentes. - Implementa retroceso exponencial y reintentos para manejar errores 429.
- Monitorea los límites de velocidad actuales a través de los encabezados de respuesta.
Ejemplo en Python:
from cognite.client import CogniteClient
from cognite.client.data_classes import InstanceList
client = CogniteClient()
# Crear un QueuedTaskRunner con un máximo de 5 tareas concurrentes
runner = client.iam.queued_task_runner(max_concurrent=5)
def fetch_instances(space, external_id):
return client.data_modeling.instances.list(space=space, external_id=[external_id])
# Encolar tareas y ejecutarlas concurrentemente
task1 = runner.submit(fetch_instances, "space1", "id1")
task2 = runner.submit(fetch_instances, "space1", "id2")
result1 = task1.result()
result2 = task2.result()
Paginación
instances.list
Al listar instancias, la API devuelve resultados paginados. Utiliza el cursor para recorrer todas las páginas:
from cognite.client import CogniteClient
client = CogniteClient()
cursor = None
all_instances = []
while True:
result = client.data_modeling.instances.list(space="mi_espacio", cursor=cursor, limit=1000)
all_instances.extend(result.data)
if result.cursor is None:
break
cursor = result.cursor
instances.query
Para consultas que devuelven muchas instancias, utiliza la paginación con nextCursor:
cursor = None
while True:
result = client.data_modeling.instances.query(
filter=...,
cursor=cursor,
limit=1000
)
# Procesar resultados
if result.next_cursor is None:
break
cursor = result.next_cursor
Operaciones de escritura por lotes
Al crear o actualizar instancias, agrupa las operaciones en lotes para reducir la cantidad de solicitudes:
from cognite.client.data_classes import InstanceWrite
lote = []
for i in range(1000):
lote.append(InstanceWrite(space="mi_espacio", external_id=f"id_{i}", ...))
if len(lote) == 100:
client.data_modeling.instances.upsert(lote)
lote = []
# Enviar el lote restante
if lote:
client.data_modeling.instances.upsert(lote)
Búsqueda vs Filtro
instances.search: Úsalo cuando necesites buscar por términos de texto en campos indexados para búsqueda. Es útil para búsquedas por palabras clave.instances.filter: Úsalo para filtrar por propiedades exactas, rangos o condiciones booleanas. Es más eficiente para consultas estructuradas.
| Operación | Caso de uso | Rendimiento |
|---|---|---|
instances.search | Búsqueda por texto flexible | Puede ser más lento en conjuntos de datos grandes |
instances.filter | Filtros exactos y estructurados | Generalmente más rápido |
QueuedTaskRunner (Semáforo)
QueuedTaskRunner es una utilidad que te permite controlar la concurrencia de las solicitudes a la API de CDF. Funciona como un semáforo, limitando el número de tareas que se ejecutan simultáneamente.
Beneficios:
- Previene la sobrecarga del sistema y errores 429.
- Mejora la fiabilidad de las aplicaciones.
- Simplifica el código de control de concurrencia.
Ejemplo en TypeScript:
import { CogniteClient } from "@cognite/sdk";
const client = new CogniteClient({ appId: "mi-app", project: "mi-proyecto" });
const runner = client.createQueuedTaskRunner({ maxConcurrent: 5 });
async function fetchInstance(space: string, externalId: string) {
return client.dataModeling.instances.list({ space, externalId: [externalId] });
}
const tasks = [
runner.submit(() => fetchInstance("space1", "id1")),
runner.submit(() => fetchInstance("space1", "id2")),
];
const results = await Promise.all(tasks);
Resumen de desencadenadores
- DMS limits: Errores relacionados con límites de la API de Data Modeling.
- 429 error: Error de límite de velocidad.
- Rate limit: Límite de velocidad genérico.
- Pagination: Necesidad de recorrer páginas de resultados.
- Cursor / nextCursor: Parámetros para la paginación.
- Batching: Agrupación de operaciones para eficiencia.
- Semaphore / QueuedTaskRunner / cdfTaskRunner: Herramientas para control de concurrencia.
- instances.search / instances.list / instances.query / instances.upsert: Operaciones específicas de la API.
- Concurrency / deadlock: Problemas de concurrencia y bloqueos.