@kapa.ai Como hago para que cargue más rápido y no demore 1 hora

version: 7.0.4

type: DeclarativeSource check: type: CheckStream stream_names: - ads_parent definitions: linked: HttpRequester: authenticator: type: BearerAuthenticator api_token: "{{ config.access_token }}" error_handler: type: DefaultErrorHandler max_retries: 15 response_filters: - type: HttpResponseFilter action: RETRY http_codes: - 400 - 403 - 429 error_message_contains: too many calls - type: HttpResponseFilter action: RETRY http_codes: - 400 - 403 - 429 error_message_contains: Application request limit reached backoff_strategies: - type: ConstantBackoffStrategy backoff_time_in_seconds: 300 request_parameters: limit: "100" fields: id,name,creative{id,image_hash,asset_feed_spec} SimpleRetriever: paginator: type: DefaultPaginator page_token_option: type: RequestOption field_name: after inject_into: request_parameter pagination_strategy: type: CursorPagination cursor_value: "{{ response.get('paging', {}).get('cursors', {}).get('after') }}" stop_condition: |- {{ 'paging' not in response or 'next' not in response.get('paging', {}) }} streams: - type: DeclarativeStream name: ads_parent retriever: type: SimpleRetriever paginator: $ref: "#/definitions/linked/SimpleRetriever/paginator" requester: type: HttpRequester url: |- https://graph.facebook.com/v25.0/act_{{ stream_slice.current_account_id }}/ads http_method: GET authenticator: $ref: "#/definitions/linked/HttpRequester/authenticator" error_handler: $ref: "#/definitions/linked/HttpRequester/error_handler" request_parameters: $ref: "#/definitions/linked/HttpRequester/request_parameters" record_selector: type: RecordSelector extractor: type: DpathExtractor field_path: - data record_filter: type: RecordFilter condition: "{{ record.get('safe_hash', '') != '' }}" transform_before_filtering: true partition_router: type: ListPartitionRouter values: "{{ config.account_ids }}" cursor_field: current_account_id primary_key: - - id schema_loader: type: InlineSchemaLoader schema: type: object $schema: http://json-schema.org/schema# properties: id: type: string name: type: string safe_hash: type: - string - "null" parent_backpack: type: - string - "null" transformations: - type: AddFields fields: - type: AddedFieldDefinition path: - safe_hash value: |- {%- set c = record.get('creative', {}) -%} {%- set afs = c.get('asset_feed_spec', {}) -%} {%- set imgs = afs.get('images', []) -%} {%- set vids = afs.get('videos', []) -%} {%- if c.get('image_hash') -%} {{- c.get('image_hash') -}} {%- elif imgs|length > 0 and imgs[0].get('hash') -%} {{- imgs[0].get('hash') -}} {%- elif vids|length > 0 and vids[0].get('thumbnail_hash') -%} {{- vids[0].get('thumbnail_hash') -}} {%- else -%} {{- '' -}} {%- endif -%} - type: AddedFieldDefinition path: - parent_backpack value: "{{ record.id }}~~{{ record.name }}~~{{ record.safe_hash }}" - type: DeclarativeStream name: ad_images_child retriever: type: SimpleRetriever requester: type: HttpRequester url: |- https://graph.facebook.com/v25.0/act_{{ stream_slice.parent_slice.current_account_id }}/adimages?hashes=["{{ stream_slice.backpack.split("~~")[2] }}"] http_method: GET authenticator: $ref: "#/definitions/linked/HttpRequester/authenticator" error_handler: $ref: "#/definitions/linked/HttpRequester/error_handler" request_parameters: fields: hash,url,width,height record_selector: type: RecordSelector extractor: type: DpathExtractor field_path: - data partition_router: type: SubstreamPartitionRouter parent_stream_configs: - type: ParentStreamConfig stream: $ref: "#/streams/0" parent_key: parent_backpack partition_field: backpack primary_key: - - hash schema_loader: type: InlineSchemaLoader schema: type: object $schema: http://json-schema.org/schema# properties: url: type: string hash: type: string ad_id: type: string width: type: integer height: type: integer ad_name: type: string transformations: - type: AddFields fields: - type: AddedFieldDefinition path: - ad_id value: "{{ stream_slice.backpack.split('~~')[0] }}" - type: AddedFieldDefinition path: - ad_name value: "{{ stream_slice.backpack.split('~~')[1] }}" spec: type: Spec connection_specification: type: object $schema: http://json-schema.org/draft-07/schema# required: - account_ids - access_token properties: account_ids: type: array items: type: string title: Facebook Ad Account IDs access_token: type: string title: Access Token airbyte_secret: true metadata: testedStreams: ads_parent: streamHash: 7e32f06c1cd8b3479a9a433d88a1456a4660213a ad_images_child: streamHash: 07845fa6b57b49325e4233235d6cb29f5b674af4 concurrency_level: type: ConcurrencyLevel default_concurrency: 1

Hay varias cosas que puedes ajustar en tu declarative source para que el sync sea más rápido, basándonos en lo que documenta Airbyte sobre paginación, concurrencia y rendimiento. 1. Aumentar el

limit

(page size) Ahora mismo estás usando:

request_parameters:
  limit: "100"
  fields: id,name,creative{id,image_hash,asset_feed_spec}

La documentación recomienda usar el tamaño de página más grande que soporte la API para reducir el número de requests. [Perf paginación] Si la API de Facebook lo permite, prueba con algo como:

request_parameters:
  limit: "500"
  fields: id,name,creative{id,image_hash,asset_feed_spec}

o el máximo que admita el endpoint sin romper límites de rate limit. 2. Subir la concurrencia Tienes:

concurrency_level:
  type: ConcurrencyLevel
  default_concurrency: 1

Con

1

solo se procesa un stream/partición a la vez. La definición de

ConcurrencyLevel

indica que este valor controla cuántos streams o particiones se procesan en paralelo, y que valores mayores permiten más paralelismo. [Concurrency ref] Puedes subirlo, por ejemplo:

concurrency_level:
  type: ConcurrencyLevel
  default_concurrency: 5
  max_concurrency: 10

(ajusta según la capacidad de tu infraestructura y lo que aguante la API de Facebook sin rate limit). 3. Revisa el backoff y los reintentos Tienes:

backoff_strategies:
  - type: ConstantBackoffStrategy
    backoff_time_in_seconds: 300
max_retries: 15

Cada vez que Facebook devuelva 400/403/429 con esos mensajes, esperas 5 minutos y puedes reintentar hasta 15 veces. Eso por sí solo puede explicar que el sync tarde cerca de 1 hora si hay varios rate limits. Si la API lo permite, reduce el

backoff_time_in_seconds

y/o

max_retries

para que, en caso de límite, no se alargue tanto. La documentación no da valores recomendados concretos, solo indica que hay que respetar los límites de la API, así que aquí tendrás que equilibrar velocidad vs. riesgo de más errores. [Speed improvements] 4. Mantener la paginación por cursor Tu configuración de paginación por cursor es correcta según la documentación:

paginator:
  type: DefaultPaginator
  page_token_option:
    type: RequestOption
    field_name: after
    inject_into: request_parameter
  pagination_strategy:
    type: CursorPagination
    cursor_value: "{{ response.get('paging', {}).get('cursors', {}).get('after') }}"
    stop_condition: |-
      {{ 'paging' not in response or 'next' not in response.get('paging', {}) }}

Esto sigue el patrón de

CursorPagination

usando

cursor_value

y

stop_condition

para dejar de pedir páginas cuando ya no hay

next

. [Cursor strategy] Aquí el principal acelerador será el

limit

más alto (punto 1). 5. Consideraciones adicionales • La documentación de rendimiento de paginación recomienda: ◦ Usar el mayor page size posible. ◦ Definir buenas condiciones de parada (ya lo haces con

stop_condition

). ◦ Probar primero con page sizes pequeños para validar y luego subir. [Perf paginación] • La documentación de “faster sync speed” indica que el hardware (CPU) y la versión de Airbyte también influyen, pero no da parámetros adicionales que puedas cambiar en el YAML. [Faster sync] En resumen, para que no demore 1 hora: 1. Sube

limit

lo más que permita la API. 2. Aumenta

default_concurrency

(y define

max_concurrency

) para procesar más particiones en paralelo. 3. Reduce el

backoff_time_in_seconds

y/o

max_retries

si los límites de la API lo permiten. 👍 0 button 👎 0 button ℹ️ Tip: Mention @kapa.ai in the thread for follow-ups or DM the app for 1-to-1 help.