airbytehq / airbytehq/PyAirbyte

s3 source stream doesn't have cursor field defined

未关闭
#575 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
主要语言
Python
星标
344
派生
77
平均合并
1 天 11 小时
30 天内合并 PR
35

描述

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:?]

贡献指南

打开贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。