airbytehq / airbytehq/PyAirbyte

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

Đang mở
#580 2 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
344
Fork
77
Merge trung bình
1 ngày 11 giờ
Pull request đã merge (30 ngày)
35

Mô tả

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()
```

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.