- 先看原数据结构和数据样例
SparkSession spark=SparkSession.builder().master("local[*]").getOrCreate();
Dataset<Row> table=spark.read().json("src/main/data/transfer_line_count.json");
table.printSchema();
table.show();root
|-- from_city_name: string (nullable = true)
|-- to_city_name: string (nullable = true)
|-- total_num: long (nullable = true)
+--------------+------------+---------+
|from_city_name|to_city_name|total_num|
+--------------+------------+---------+
| 新乡市| 上海市| 2|
| 南昌市| 天津市| 3|
| 抚州市| 温州市| 36|
| 娄底市| 中山市| 1|
| 广州市| 孝感市| 3|
| 天门市| 岳阳市| 1|
| 长春市| 沈阳市| 6|
| 成都市| 眉山市| 10|
| 广州市| 怀集县| 2|
| 北海市| 广州市| 8|- 业务需求
from_city_name开始车票起始城市,to_city_name表示到达城市,
都是group by去重的,total_num表示改线路订单总数.
现在有这样需求,统计每行线路数据到占所有订单总数(sum(total_num))的百分比.
比如在数据样本中,订单总数为
table.agg(sum("total_num")).show();+--------------+
|sum(total_num)|
+--------------+
| 20000|
+--------------+比如这个例子中
新乡市->上海市 所占的百分比为 2/20000*100
代码如下
Dataset<Row> result=table.withColumn("percent",format_number(col("total_num").divide( sum("total_num").over()).multiply(100),5));
result.show();
运行结果如下:
+--------------+------------+---------+-------+
|from_city_name|to_city_name|total_num|percent|
+--------------+------------+---------+-------+
| 新乡市| 上海市| 2|0.01000|
| 南昌市| 天津市| 3|0.01500|
| 抚州市| 温州市| 36|0.18000|
| 娄底市| 中山市| 1|0.00500|
| 广州市| 孝感市| 3|0.01500|
| 天门市| 岳阳市| 1|0.00500|
| 长春市| 沈阳市| 6|0.03000|
| 成都市| 眉山市| 10|0.05000|
| 广州市| 怀集县| 2|0.01000|
| 北海市| 广州市| 8|0.04000|
| 商丘市| 曹县| 2|0.01000|
| 册亨县| 罗平县| 1|0.00500|
| 文水县| 鄂尔多斯市| 3|0.01500|
| 赣州市| 广州市| 3|0.01500|
| 枣庄市| 邳州市| 1|0.00500|
| 许昌市| 漯河市| 6|0.03000|
| 柳州市| 南宁市| 3|0.01500|
| 娄底市| 珠海市| 1|0.00500|
| 邵阳市| 永州市| 1|0.00500|
| 山阴县| 太原市| 1|0.00500|
+--------------+------------+---------+-------+
only showing top 20 rows