BigQuery с PySpark перезаписывает идентификатор проекта

Question

Я использую BigQuery и Datapro c в Google Cloud. Оба находятся в одном проекте, назовем его «проект-123». Я использую Composer (Airflow) для запуска своего кода.

У меня есть простой python скрипт test_script.py, который использует pyspark для получения данных чтения из таблицы в наборе данных bigquery publi c :

if __name__ == "__main__":
    # Create Spark Cluster
    try:
        spark = SparkSession.builder.appName("test_script").getOrCreate()
        log.info("Created a SparkSession")
    except ValueError:
        warnings.warn("SparkSession already exists in this scope")

    df = (
        spark.read.format("bigquery")
        .option("project", "project-123")
        .option("dataset", "bigquery-public-data")
        .option("table", "crypto_bitcoin.outputs")
        .load()
    )

Я запускаю сценарий, используя DataProcPySparkOperator в потоке воздуха:

    # This task corresponds to the ""
    test_script_task = DataProcPySparkOperator(
        task_id="test_script",
        main="./test_script.py",
        cluster_name="test_script_cluster",
        arguments=[],

        # Since we are using bigquery, we need to explicity add the connector jar
        dataproc_pyspark_jars="gs://spark-lib/bigquery/spark-bigquery-latest.jar",
    )

Однако при каждой попытке я получаю следующую ошибку:

Invalid project ID '/tmp/test_script_20200304_407da59b/test_script.py'. Project IDs must contain 6-63 lowercase letters, digits, or dashes. Some project IDs also include domain name separated by a colon. IDs must start with a letter and may not end with a dash.

Откуда берется этот идентификатор проекта? Это явно не перезаписывается моим .option("project", "project-123"). Я предполагаю, что Composer хранит мой сценарий искровой работы в местоположении /tmp/test_script_20200304_407da59b/test_script.py. Если это так, как я могу перезаписать идентификатор проекта?

Любая помощь очень ценится

Answer 1

Боюсь, вы смешиваете параметры. project - это проект, к которому принадлежит таблица, а bigquery-public-data - это проект, а не набор данных. Пожалуйста, попробуйте следующий звонок:

df = (
        spark.read.format("bigquery")
        .option("parentProject", "project-123")
        .option("project", "bigquery-public-data")
        .option("table", "crypto_bitcoin.outputs")
        .load()
    )