Вычисление асимметрии с использованием spark.sql и Cloudant

Question

У меня проблемы со следующим фрагментом кода:

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

Кто-нибудь знает, как это исправить?

Answer 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 не решит проблему, вам может потребоваться перейти на план обслуживания с более высоким порогом.