1

我对以下代码有问题：

def skewTemperature(cloudantdata,spark):
    return spark.sql("""SELECT (1/count(temperature)) * (sum(POW(temperature-%s,3))/pow(%s,3)) as skew from washing""" %(meanTemperature(cloudantdata,spark),sdTemperature(cloudantdata,spark))).first().skew

meanTemperature并且sdTemperature工作正常，但使用上述查询我收到以下错误：

Py4JJavaError: An error occurred while calling o2849.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 315.0 failed 10 times, most recent failure: Lost task 3.9 in stage 315.0 (TID 1532, yp-spark-dal09-env5-0045): java.lang.RuntimeException: Database washing request error: {"error":"too_many_requests","reason":"You've exceeded your current limit of 5 requests per second for query class. Please try later.","class":"query","rate":5

有人知道如何解决这个问题吗？

4

1 回答 1

1

该错误表明您超出了查询类的 Cloudant API 调用阈值，对于您正在使用的服务计划，该阈值似乎是 5/秒。一种可能的解决方案是通过定义配置属性来限制分区的数量jsonstore.rdd.partitions，如以下 Spark 2 示例所示：

spark = SparkSession\    
        .builder\    
        .appName("Cloudant Spark SQL Example in Python using dataframes")\
        .config("cloudant.host","ACCOUNT.cloudant.com")\     
        .config("cloudant.username", "USERNAME")\    
        .config("cloudant.password","PASSWORD")\    
        .config("jsonstore.rdd.partitions", 5)\    
        .getOrCreate()

如果错误仍然存​​在，请从 5 开始并逐步降低到 1。此设置基本上限制了将发送到 Cloudant 的并发请求的数量。如果设置为 1 不能解决问题，您可能必须考虑升级到具有更大阈值的服务计划。

于 2017-09-06T18:32:42.003 回答