如何在 DataFrame 中跨组使用 QuantileDiscretizer？

我有一个包含以下列的数据框。

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分位数离散化器 http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.ml.feature.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然后一个接一个地把它扔进垃圾桶？

我一无所知QuantileDiscretizer，但认为您最关心的是要应用的数据集QuantileDiscretizer到。我认为您想弄清楚如何将输入数据集分割成更小的数据集show_name（你说有 192 个不同的show_name在输入数据集中）。

解决方案1：对Parquet数据集进行分区

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

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

这将使你的情况变得微不足道，因为分割是在编写时完成的（又名不再是我的问题).

See 如何在 Spark 2.1 中保存分区的 parquet 文件？ https://stackoverflow.com/q/43731679/1305344

解决方案 2：Scala 的未来

给定您的迭代解决方案，您可以将每次迭代包装成Future https://www.scala-lang.org/api/current/scala/concurrent/Future.html您将提交并行处理。

Spark SQL 的Spark会话 http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.SparkSession（和 Spark Core 的SparkContext http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext) 是线程安全的。

解决方案 3：数据集filter and union运营商

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

假设您有一个包含 1.33 亿行的大型 Parquet 文件，我首先会为每个文件构建 192 个数据集show_name using filter运算符（就像您构建show_rdd这是反对这个名字的，因为它是DataFrame not RDD) and union（再次像你一样）。

See 数据集API http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Dataset.

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

我认为这是可行的，但我自己没有检查过。

您可以使用窗口函数（请参阅窗口规范 http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.expressions.WindowSpec和列的over http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Column@over():org.apache.spark.sql.Column操作员）。

窗口函数将为您提供分区（窗口），同时over would somehow apply QuantileDiscretizer到窗口/分区。然而，这需要“解构”QuantileDiscretizer进入一个Estimator训练模型并somehow再次将结果模型拟合到窗口。

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