PyAirbyte+Airflow同步Google Sheet至PostgreSQL为空问题求助

我用PyAirbyte结合Airflow将Google Sheet数据同步到PostgreSQL表，代码执行无报错，但生成的表一直是空的。已经确认服务账号具备必要权限，怀疑可能漏掉了Airbyte库或source-google-sheets版本相关的额外步骤。

我的代码 #

from airflow.decorators import task

@task.external_python(python="/venv_airbyte/bin/python")
def extract():
    import airbyte as ab
    def get_credentials():
        with open(PATH_to_KEY) as f:
            return f.read()

    #VARIABLES
    PATH_to_KEY = "my_service_account.json"
    SPREADSHEET_ID = "1pYS24BwqRl5hIhVAjo1QFDsupTx67nYCTCTIF-PcVoY"
    DB_HOST="localhost"
    DB_PORT=5423
    DB_USR=""
    DB_PWD=""
    DB_NAME=""


    credentials = get_credentials() # 获取字符串格式的JSON服务账号凭证
    source = ab.get_source(
        "source-google-sheets",
        pip_url="airbyte-source-google-sheets unidecode",
        install_if_missing=True,
        config={
            "spreadsheet_id": SPREADSHEET_ID,
            "credentials": {
                "auth_type": "Service",
                "service_account_info": credentials,
            },
        },
    )
    source.select_all_streams()

    pg_cache = ab.caches.PostgresCache(
            host=DB_HOST,
            port=5432,
            username=DB_USR,
            password=DB_PWD,
            database=DB_NAME,
            schema_name="raw"
    )
    read_result = source.read(cache=pg_cache)

    # 调试：查看read_result结构和流信息
    print("Read Result Streams:", read_result.streams)

电子表格示例 #

使用的版本信息 #

名称：airbyte 版本：0.24.2
概述：PyAirbyte
依赖：airbyte-api, airbyte-cdk, airbyte-protocol-models-pdv2, click, duckdb, duckdb-engine, google-auth, google-cloud-bigquery, google-cloud-bigquery-storage, google-cloud-secret-manager, jsonschema, orjson, overrides, pandas, psycopg, psycopg2-binary, pyarrow, pydantic, pydantic-core, python-dotenv, python-ulid, pytz, pyyaml, requests, rich, snowflake-connector-python, snowflake-sqlalchemy, sqlalchemy, sqlalchemy-bigquery, structlog, typing-extensions, uuid7

内容的提问来源于stack exchange，提问作者dev_community