Apache Spark GCS 连接器的速率限制

Question

我在带有 Google Cloud Storage 连接器（而不是 HDFS，如 recommended）的 Google Compute Engine 集群上使用 Spark，并收到很多“速率限制”错误，如下所示：

java.io.IOException: Error inserting: bucket: *****, object: *****
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.wrapException(GoogleCloudStorageImpl.java:1600)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl$3.run(GoogleCloudStorageImpl.java:475)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 429 Too Many Requests
{
  "code" : 429,
  "errors" : [ {
    "domain" : "usageLimits",
    "message" : "The total number of changes to the object ***** exceeds the rate limit. Please reduce the rate of create, update, and delete requests.",
    "reason" : "rateLimitExceeded"
  } ],
  "message" : "The total number of changes to the object ***** exceeds the rate limit. Please reduce the rate of create, update, and delete requests."
}
  at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:145)
  at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:113)
  at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:40)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:432)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:352)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:469)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl$3.run(GoogleCloudStorageImpl.java:472)
  ... 3 more

• 有人知道有什么解决办法吗？

• 有没有办法控制Spark的读写速率？

• 有没有办法提高我的 Google 项目的速率限制？

• 有没有办法使用本地硬盘存储没有的临时文件 与其他奴隶共享？

谢谢！

Answer 1

不幸的是，将 GCS 设置为 DEFAULT_FS 时，无论将其用于中间目录还是用于最终输入/输出目录，都会以高速率创建目录对象。尤其是使用 GCS 作为最终输出目录时，很难应用任何 Spark 端的解决方法来降低冗余目录创建请求的速率。

好消息是这些目录请求中的大多数确实是多余的，只是因为系统习惯于能够本质上“mkdir -p”，并且如果目录已经存在则廉价地返回true。在我们的例子中，可以在 GCS 连接器端通过捕获这些错误来修复它，然后检查该目录是否确实是由其他工作人员在竞争条件下创建的。

现在应该用https://github.com/GoogleCloudPlatform/bigdata-interop/commit/141b1efab9ef23b6b5f5910d8206fcbc228d2ed7 修复这个问题

要测试，只需运行：

git clone https://github.com/GoogleCloudPlatform/bigdata-interop.git
cd bigdata-interop
mvn -P hadoop1 package
# Or or Hadoop 2
mvn -P hadoop2 package

您应该会找到可用的文件“gcs/target/gcs-connector-*-shaded.jar”。要将其插入 bdutil，只需 gsutil cp gcs/target/gcs-connector-*shaded.jar gs://<your-bucket>/some-path/，然后为 Hadoop 1 编辑 bdutil/bdutil_env.sh 或 bdutil/hadoop2_env.sh 进行更改：

GCS_CONNECTOR_JAR='https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-1.4.1-hadoop2.jar'

改为指向您的gs://<your-bucket>/some-path/ 路径； bdutil 会自动检测到您正在使用以 gs:// 为前缀的 URI，并将在部署期间执行正确的操作。

如果它为您解决了问题，请告诉我们！

Answer 2

您是否尝试设置 spark.local.dir 配置参数并将该 tmp 空间的磁盘（首选 SSD）附加到您的 Google Compute Engine 实例？

https://spark.apache.org/docs/1.2.0/configuration.html

您无法更改项目的速率限制，一旦达到限制，您将不得不使用回退算法。由于您提到大多数读取/写入都是针对 tmp 文件的，因此请尝试将 Spark 配置为使用本地磁盘。