airbytehq / airbytehq/PyAirbyte

Shopify<>GCS (State Cache: Postgres): Frozen After Completion

Abierto
#580 2 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
344
Forks
77
Merge medio
1 d 11 h
PR fusionados (30 d)
35

Descripción

I'm running PyAirbyte to sync a Shopify Source to a Google Cloud Storage Destination using a Postgres Cache as a State Cache.
The process runs smoothly and takes around 15 minutes in total, as depicted by the image below:

Image

The odd thing is, the process never completes after this. It just gets stuck and the Python process never ends.

Below are the code and the logs:
- Log:[ Shopify Source](https://gist.github.com/Guilherme-B/d56c025d57feee195fd997902eebd936)
- Log: [GCS Destination](https://gist.github.com/Guilherme-B/ff428eeb2287c3ab92403572978055aa)

Code:

```
from __future__ import annotations

import datetime

import airbyte as ab
from airbyte.caches import PostgresCache

def get_shopify_source() -> ab.Source:
return ab.get_source(
"source-shopify",
config={
"shop": "",
"credentials": {
"auth_method": "api_password",
"api_password": "",
},
},
# docker_image=True,
streams=[
"blogs",
"collections",
"collects",
"countries",
"custom_collections",
"customer_address",
"customers",
"discount_codes",
"draft_orders",
"fulfillment_orders",
"inventory_items",
"inventory_levels",
"locations",
"metafield_collections",
"metafield_customers",
"metafield_products",
"metafield_shops",
"metafield_smart_collections",
"order_risks",
"orders",
"pages",
"price_rules",
"product_images",
"product_variants",
"products",
# "products_graph_ql",
"shop",
"smart_collections",
"tender_transactions",
"transactions",
],
)

def get_gcs_destination() -> ab.Destination:
# Destination configuration
gcs_config = {
"gcs_bucket_name": "",
"gcs_bucket_path": "",
"gcs_bucket_region": "",
"credential": {
"credential_type": "HMAC_KEY",
"hmac_key_access_id": "",
"hmac_key_secret": "",
},
"format": {
"format_type": "Parquet",
"compression_codec": "UNCOMPRESSED",
"block_size_mb": 128,
},
}

destination = ab.get_destination(
name="destination-gcs",
config=gcs_config,
install_if_missing=True,
docker_image=False,
)

return destination

def main() -> None:
"""Test writing from the source to the destination."""
source = get_shopify_source()
source.check()
destination = get_gcs_destination()
destination.check()

# Define a Postgres Cache and pass the necessary configuration
pg_state_cache = PostgresCache(
host="localhost",
port=5432,
username="X",
password="X",
database="pyairbyte_demo",
schema_name="airbyte_internal",
)

destination._name = "client_" + destination.name
source._name = "client_" + source.name

write_result: ab.WriteResult = destination.write(
source_data=source,
state_cache=pg_state_cache,
cache=False,
)

print(
f"Completed writing {write_result.processed_records:,} records "
f"to destination at {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}.",
)

if __name__ == "__main__":
main()
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.