scala - 如何在 DataFrame 中跨组使用 QuantileDiscretizer？

Question

我有一个包含以下列的 DataFrame。

scala> show_times.printSchema
root
 |-- account: string (nullable = true)
 |-- channel: string (nullable = true)
 |-- show_name: string (nullable = true)
 |-- total_time_watched: integer (nullable = true)

这是关于客户观看特定节目多少次的数据。我应该根据观看的总时间对每个节目的客户进行分类。

该数据集共有 1.33 亿行，有 192 个不同的 show_names 。

对于每个单独的节目，我应该将客户分为 3 个类别 (1,2,3)。

我使用 Spark MLlib 的 QuantileDiscretizer

目前，我遍历每个节目并按顺序运行 QuantileDiscretizer，如下面的代码所示。



我最终想要的是以下示例输入以获取示例输出。

样本输入:

account,channel,show_name,total_time_watched
acct1,ESPN,show1,200
acct2,ESPN,show1,250
acct3,ESPN,show1,800
acct4,ESPN,show1,850
acct5,ESPN,show1,1300
acct6,ESPN,show1,1320
acct1,ESPN,show2,200
acct2,ESPN,show2,250
acct3,ESPN,show2,800
acct4,ESPN,show2,850
acct5,ESPN,show2,1300
acct6,ESPN,show2,1320

示例输出:

account,channel,show_name,total_time_watched,Time_watched_bin
acct1,ESPN,show1,200,1
acct2,ESPN,show1,250,1
acct3,ESPN,show1,800,2
acct4,ESPN,show1,850,2
acct5,ESPN,show1,1300,3
acct6,ESPN,show1,1320,3
acct1,ESPN,show2,200,1
acct2,ESPN,show2,250,1
acct3,ESPN,show2,800,2
acct4,ESPN,show2,850,2
acct5,ESPN,show2,1300,3
acct6,ESPN,show2,1320,3

有没有一种更有效和分布式的方法来使用一些类似 groupBy 的操作而不是循环遍历每个 show_name 并将其一个接一个地装箱？

Answer 1

我对 QuantileDiscretizer 一无所知，但认为您最关心的是将 QuantileDiscretizer 应用到的数据集。我想您想弄清楚如何将输入数据集拆分为每个 show_name 的较小数据集(您说输入数据集中有 192 个不同的 show_name)。

解决方案 1:分区 Parquet 数据集

我注意到您使用 parquet 作为输入格式。我对格式的理解非常有限，但我注意到人们正在使用某种分区方案将大数据集拆分成更小的块，然后他们可以处理他们喜欢的任何内容(根据某些分区方案)。

在您的情况下，分区方案可能包括 show_name 。

这将使您的案例变得微不足道，因为拆分是在写作时完成的(也不再是我的问题)。

见 How to save a partitioned parquet file in Spark 2.1?

解决方案 2:Scala 的 future

给定您的迭代解决方案，您可以将每次迭代都包装到一个 Future 中，然后提交以并行处理。

Spark SQL 的 SparkSession (和 Spark Core 的 SparkContext )是线程安全的。

方案三:Dataset 的 filter 和 union 操作符

在遵循此解决方案之前，我会三思而后行，因为它会给您的肩膀带来负担，我认为这可以通过解决方案 1 轻松解决。

鉴于您有一个 1.33 亿行的大型 Parquet 文件，我首先使用 show_name 运算符为每个 filter 构建 192 个数据集(就像您构建 show_rdd 那样，它与名称相反，因为它是 DataFrame 而不是 RDD )和 union (再次像你一样)。

见 Dataset API 。

解决方案 4:使用窗口函数

这是我认为可以工作的东西，但我自己没有检查过。

您可以使用窗口函数(请参阅 WindowSpec 和 Column 的 over 运算符)。

窗口函数会给你分区(窗口)，而 over 会以某种方式将 QuantileDiscretizer 应用于窗口/分区。然而，这需要将 QuantileDiscretizer “解构”成 Estimator 以训练模型并以某种方式再次将结果模型拟合到窗口中。

我认为这是可行的，但我自己没有做过。对不起。