apache-spark - Google Dataproc Pyspark - BigQuery 连接器 super 慢

Question

我尝试使用 Spark Bigquery connector 将 10 万行写入 Bigquery .

我的行由 2 个大字符串(每个大约 200-250 个单词)、许多单个单词字符串和一些 dict 类型的数据(最大深度 2，里面的数据很少)组成，总共大约 35 个字段。

我对字符串做了一些处理，对于 10 万行，这几乎是即时的，但是当涉及到将数据写入 BQ 时，问题就出现了。

我使用一个由 5 个工作节点组成的 Spark 集群，每个工作节点有 32GB RAM、8 个 vCPU 和 500GB SSD，40 个 vCPU 总共有 160GB RAM。即使有这些规范，将 10 万行写入 BQ 也需要 50 分钟。我调查了一下，因为我使用的是间接写入，所以数据首先写入 GCS，然后由 BQ 读取。读取作业大约需要。 20 秒，这意味着仅 10 万行的 GCS 写入操作就需要 50 分钟。

这不可能是正常行为，因为即使我在本地使用我的家用计算机和 pandas 例如运行相同的写入操作，它所花费的时间也会比这少得多。

我的 Spark Session 是这样初始化的:

spark = SparkSession \
    .builder \
    .appName('extract-skills') \
    .config('spark.jars.packages', 'com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.26.0,com.johnsnowlabs.nlp:spark-nlp_2.12:4.1.0') \
    .config('spark.executor.memory', '25g') \
    .config('spark.executor.cores', '8') \
    .config('spark.driver.memory', '12g') \
    .config('spark.executor.instances', '5') \
    .config("spark.driver.maxResultSize", "0") \
    .config("spark.kryoserializer.buffer.max", "2000M")\
    .getOrCreate()

我的写法如下:

result. \
    write.format('bigquery') \
        .mode("overwrite") \
        .option("writeMethod", "indirect") \
        .option("temporaryGcsBucket","my_bucket") \
        .option('table', 'my_project.my_dataset.my_table') \
        .save()

我在这里忘记了什么吗？我找不到瓶颈，将 writeMethod 设置为 direct 是不可能的，因为我需要在分区表中写入。

Answer 1

使用直接写入应该更快，变化非常小:

result. \
    write.format('bigquery') \
        .mode("overwrite") \
        .option("writeMethod", "direct") \
        .option('table', 'my_project.my_dataset.my_table') \
        .save()

此外，请检查 DataFrame 是否正确分区。如果单个分区比其他分区大得多，那么您的资源使用效率很低。