Pyspark 代码在多次运行时在同一组数据上的不同输出答案

【问题标题】：Different Output of Pyspark code on same set of data while multiple runsPyspark 代码在多次运行时在同一组数据上的不同输出
【发布时间】：2019-05-20 07:16:37
【问题描述】：

我已经编写了一个 pyspark 函数，但是当我多次运行时，它每次都会给我不同的输出在同一组输入数据上。

-pyspark函数

def give_percentile(plat,metrics,perc):
df_perc = df_final.filter(df_final.platform.like('%' + plat + '%'))
df_perc = df_perc.filter(df_final[metrics]!=0)
percentile_val = df_perc.approxQuantile(metrics, [perc], 0.05)
if len(percentile_val)>0:
    percentile_val = float(percentile_val[0])
else:
    percentile_val = float(0)
return percentile_val

调用函数-

   df_agg = sqlContext.createDataFrame([Row(platform='iOS',
                                        percentile_page_load_50=give_percentile("iOS","page_load",0.5),
                                        percentile_time_diff_50=give_percentile("iOS","session_duration",0.5)),
                                        Row(platform='Android',
                                        percentile_page_load_50=give_percentile("Android","page_load",0.5),
                                        percentile_time_diff_50=give_percentile("Android","session_duration",0.5)),
                                        Row(platform='Web',
                                        percentile_page_load_50=give_percentile("Web","page_load",0.5),
                                        percentile_time_diff_50=give_percentile("Web","session_duration",0.5)))

Spark 作业提交：-

    spark-submit --deploy-mode cluster  --executor-cores 4 --executor-memory 12G --driver-cores 4 --driver-memory 12G --conf spark.driver.extraJavaOptions=-Duser.timezone=UTC --conf spark.executor.extraJavaOptions=-Duser.timezone=UTC "path"

我们以 parquet 文件格式存储 pyspark 代码的输出，并在其之上创建 impala 表，如下所示：

1.select a.percentile_page_load_50,a.percentile_time_diff_50 from Tablename1 a where a.platform ='Colvalue' and a.dt ='20190501' 限制 5; 表记录数= 22093826

输出 =0.62400001287460327 0.35100001096725464

2.select a.percentile_page_load_50,a.percentile_time_diff_50 from Tablename2 a where a.platform ='Colvalue' and a.dt ='20190501' 限制 5; 表记录数= 22093826

输出=0.61500000953674316 0.28499999642372131 3.select a.percentile_page_load_50,a.percentile_time_diff_50 from Tablename3 a where a.platform ='Colvalue' and a.dt ='20190501' 限制 5; 表记录数= 22093826

输出= 0.61799997091293335
0.27799999713897705

现在这里的 Tablename1、Tablename2 和 Tablename3 是在同一组输入数据上多次运行 pyspark 代码的输出。但值仍然不同，因为我们的 pyspark 代码在集群模式/分布式模式下运行。当我们检查样本数据时独立模式其值不变。那么你能不能帮我看看上面的函数代码有什么问题或任何其他集群问题？

【问题讨论】：

您将 approxQuantile 函数的允许误差设置为 0.05，这意味着它仅在以下范围内具有确定性：“如果 DataFrame 有 N 个元素，并且如果我们以概率 p 请求分位数直到错误错误，那么该算法将从 DataFrame 返回一个样本 x，以便 x 的确切秩接近 (p * N)。” Link
@cronoik 你应该把它写成答案

标签： apache-spark hadoop pyspark

【解决方案1】：

函数 approxQuantile 根据给定的 relativeError 为您提供近似解。您将 approxQuantile 函数的允许 relativeError 设置为 0.05，这意味着它仅在以下范围内是确定性的：

"如果DataFrame有N个元素并且如果我们以概率p请求分位数直到error err，那么算法将从DataFrame返回一个样本x，使得x的确切秩是接近 (p * N)。” （我已经强调了为什么你会得到不同的结果）。

如果您需要精确的分位数，您必须将 relativeError 设置为 0.0，但这也会增加运行时间。更多信息可以在documentation找到。

【讨论】：