airbytehq / airbytehq/PyAirbyte

s3 source stream doesn't have cursor field defined

Đang mở
#575 0 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 am trying out PyAirbyte and have configured the below script with credentials that work fine in airbyte cloud. I'm passing all of the config spec checks, but I keep getting an error. The logs appear to show that the streams cursor is not defined, but i don't see any place to specify the cursor in the config schema and in airbyte cloud it appears the cursor is implicilty the modified date associated with the file in the s3 bucket.

Here is my script (with parts redacted):
```
import airbyte as ab

source = ab.get_source("source-s3")
source.set_config(
{
"bucket": "REDACTED",
"aws_access_key_id": "REDACTED",
"aws_secret_access_key": "REDACTED",
"streams": [
{
"name": "all data",
"globs": ["**/REDACTED*.csv"],
"start_date": "2020-01-01T00:00:00.000000Z",
"format": {
"filetype": "csv",
},
"primary_key": "REDACTED",
}
],
}
)
source.check()
# source.print_config_spec(output_file="./source.yml")

destination = ab.get_destination("destination-mysql")
destination.set_config(
{
"database": "REDACTED",
"host": "REDACTED",
"password": "REDACTED",
"port": 3306,
"username": "REDACTED",
}
)
destination.check()
# destination.print_config_spec(output_file="./destination.yml")
destination.write(source)
```

and here is the error I am getting in the logfiles:

2025-01-15 09:06:31 - INFO - ERROR main i.a.c.i.b.s.SshWrappedDestination(getSerializedMessageConsumer):136 Exception occurred while getting the delegate consumer, closing SSH tunnel java.lang.NullPointerException: Cannot invoke "java.util.List.size()" because the return value of "io.airbyte.protocol.models.v0.ConfiguredAirbyteStream.getCursorField()" is null
at io.airbyte.integrations.base.destination.typing_deduping.CatalogParser.toStreamConfig(CatalogParser.kt:134) ~[airbyte-cdk-typing-deduping-0.33.0.jar:?]
at io.airbyte.integrations.base.destination.typing_deduping.CatalogParser.parseCatalog(CatalogParser.kt:29) ~[airbyte-cdk-typing-deduping-0.33.0.jar:?]
at io.airbyte.cdk.integrations.destination.jdbc.AbstractJdbcDestination.getV2MessageConsumer(AbstractJdbcDestination.kt:294) ~[airbyte-cdk-db-destinations-0.33.0.jar:?]
at io.airbyte.cdk.integrations.destination.jdbc.AbstractJdbcDestination.getSerializedMessageConsumer(AbstractJdbcDestination.kt:264) ~[airbyte-cdk-db-destinations-0.33.0.jar:?]
at io.airbyte.cdk.integrations.base.ssh.SshWrappedDestination.getSerializedMessageConsumer(SshWrappedDestination.kt:130) [airbyte-cdk-core-0.33.0.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:208) [airbyte-cdk-core-0.33.0.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:116) [airbyte-cdk-core-0.33.0.jar:?]
at io.airbyte.integrations.destination.mysql.MySQLDestination$Companion.main(MySQLDestination.kt:234) [io.airbyte.airbyte-integrations.connectors-destination-mysql.jar:?]
at io.airbyte.integrations.destination.mysql.MySQLDestination.main(MySQLDestination.kt) [io.airbyte.airbyte-integrations.connectors-destination-mysql.jar:?]

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.