SparkSQL 加入父/子数据集

Posted 2023-04-17

【中文标题】SparkSQL 加入父/子数据集

【英文标题】：SparkSQL joining parent/child datasets

【发布时间】：2017-10-12 17:42:59

【问题描述】：

我正在使用 SparkSQL 2.2.0 从 Cassandra 加载数据并将其索引到 Elasticsearch。我拥有的数据由客户（第一张表people）和订单（第二张表orders）组成。 表订单有一个列 person_id 指向相应的客户。 我需要查询（并稍后在 Elasticsearch 中编制索引）people 表和 orders，以便我可以为每个客户获取她购买的订单数量。 我想出的最简单的方法是将这两个表读入org.apache.spark.sql.Dataset<Row>s 并在person_id 列上进行join。然后我groupBy(person_id). 这给了我一个包含两列的数据集：person_id 和count，我必须将它们与people 表结合起来，这样我就可以对其他人的数据进行计数。

Dataset<Row> peopleWithOrders = people.join(orders, people.col("id").equalTo(orders.col("person_id")), "left_outer");

Dataset<Row> peopleOrdersCounts = peopleWithOrders.groupBy("id").count().withColumnRenamed("id", "personId");

Dataset<Row> peopleWithOrderCounts = people.join(personsOrdersCounts, people.col("id").equalTo(peopleOrdersCounts.col("personId")), "left_outer")
            .withColumnRenamed("count", "nbrOfOrders")
            .select("id", "name", "birthDate", "nbrOfOrders");

people 表有 1_000_000 行，orders 有一个 2_500_000。每个客户有 2 或 3 个订单。 我正在使用带有 2.2 GHz Intel Core i7 处理器和 16 GB 1600 MHz DDR3 内存的 MAC Book pro。所有 Cassandra、Spark 2.2 master 和（单个）worker 都在同一台机器上。 这 3 个连接需要 15 到 20 秒。 我的问题是：是否有一些性能提升空间。 Window Aggregate Functions 是否有好处，因为我在日志中看到了 ShuffleMapTask。

提前致谢

【参考方案1】：

我认为第一步是不必要的。你可以这样做：

Dataset<Row> peopleOrdersCounts = orders.groupBy("person_id").count();

Dataset<Row> peopleWithOrderCounts = people.join(personsOrdersCounts, people.col("id").equalTo(peopleOrdersCounts.col("personId")), "left_outer")
            .withColumnRenamed("count", "nbrOfOrders")
            .select("id", "name", "birthDate", "nbrOfOrders");

我希望这会有所帮助。

【讨论】：

是的，这是真的。我的错。但这仍然“相对较慢”（ab 16s）。我想知道“窗口聚合函数”是否有帮助，或者这是正常的方法

据我所知，这是这样做的。特别是在“分组依据”的情况下。您可以查看用户定义的聚合函数（UDAF），但即使是针对特定情况的。是否有任何其他操作可能会减慢这一速度？