使用 PySpark 覆盖项目 ID 的 BigQuery答案

【问题标题】：BigQuery with PySpark overwriting project ID使用 PySpark 覆盖项目 ID 的 BigQuery
【发布时间】：2020-03-05 22:51:46
【问题描述】：

我在 Google Cloud 中使用 BigQuery 和 Dataproc。两者都在同一个项目中，我们称之为“project-123”。我使用 Composer (Airflow) 来运行我的代码。

我有一个简单的 Python 脚本 test_script.py，它使用 pyspark 从 bigquery 公共数据集中的表中读取数据：

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.

这个项目 ID 来自哪里？它显然没有被我的.option("project", "project-123") 覆盖。我的猜测是 Composer 将我的 Spark 作业脚本存储在位置 /tmp/test_script_20200304_407da59b/test_script.py。如果是这样，我该如何覆盖项目 ID？

非常感谢任何帮助

【问题讨论】：

标签： python pyspark google-bigquery google-cloud-dataproc google-cloud-composer

【解决方案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()
    )

【讨论】：