Spark 计算总销量
题目：
某电商平台存储了所有商品的销售数据，平台希望能够找到销量最好的前 N 个商品。通过分析销售记录，帮助平台决策哪些商品需要更多的推广资源。
假设你得到了一个商品销售记录的文本文件
product_id, product_name, quantity, sale_date
1, "Smartphone", 10, "2024-11-01"
2, "Laptop", 5, "2024-11-02"
3, "T-Shirt", 25, "2024-11-03"
4, "Smartwatch", 8, "2024-11-04"
5, "Headphones", 12, "2024-11-05"
1, "Smartphone", 15, "2024-11-06"
2, "Laptop", 10, "2024-11-07"
3, "T-Shirt", 10, "2024-11-08"
各字段含义：
product_id: 商品ID
product_name: 商品名称
quantity: 销售数量
sale_date: 销售日期
任务：
计算总销量：计算每个商品的总销量，输出如下。
product_id product_name total_sales
1 Smartphone 25
2 Laptop 15
3 T-Shirt 35
4 Smartwatch 8
5 Headphones 12
找出销量最高的前 N 个商品：根据计算出的销量，找出前 N 个销售量最多的商品，N 由用户输入。N=3时输出如下：
product_id product_name total_sales
3 T-Shirt 35
1 Smartphone 25
2 Laptop 15
运行
- 在桌面创建文件
buy_count.txt,输入文本内容 - Java代码
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.Function2;
import scala.Serializable;
import scala.Tuple2;
import java.util.Scanner;public class Test02 {/** Serializable* 标记一个类可以被序列化，* 即可以将其状态转换为字节流，* 以便进行持久化存储或在网络上传输* */static class Product implements Serializable{int product_id;String product_name;int quantity;@Overridepublic String toString() {return String.format("%-10s %-20s %-10s", product_id, product_name, quantity);}}public static void main(String[] args) {// 文件路径// 获取用户的主目录并构建绝对路径String userHome = System.getProperty("user.home");String logFile = "file://" + userHome + "/Desktop/spark_test.txt";
// String logFile = "file:///Desktop/spark_test.txt";// SparkConf 对象// setMaster("local")表示应用程序将在本地模式下运行// setAppName("SimpleApp")设置了应用程序的名称为SimpleAppSparkConf conf=new SparkConf().setMaster("local").setAppName("SimpleApp");// JavaSparkContext对象，它是与Spark交互的主要入口点。它接收前面创建的SparkConf对象作为参数JavaSparkContext sc=new JavaSparkContext(conf);// sc.textFile(logFile)加载文本文件内容// .cache()方法会将此RDD缓存起来以便后续重复使用时能更快访问JavaRDD<String> linesRDD = sc.textFile(logFile).cache();/** 按商品分组* JavaPairRDD 键值对* PairFunction用于定义将输入对象转换为键值对的逻辑* filter 方法对linesRDD中的每一行执行过滤(删除标题行)* mapToPair 会对每一行进行处理，生成键值对* 以product_name做键，Product对象做值* */JavaPairRDD<Integer, Product> productRDD = linesRDD.filter(new Function<String, Boolean>() {public Boolean call(String line) {return !line.contains("product_id");}}).mapToPair(new PairFunction<String, Integer, Product>(){@Overridepublic Tuple2<Integer, Product> call(String line) throws Exception {String[] fields = line.split(", ");Product product = new Product();product.product_id = Integer.parseInt(fields[0]);product.product_name = fields[1].replace("\"", "");product.quantity = Integer.parseInt(fields[2]);return new Tuple2<Integer, Product>(product.product_id, product);}});System.out.printf("%-10s %-20s %-10s%n", "product_id", "product_name", "total_sales");productRDD.foreach(tuple -> {Product value = tuple._2;System.out.println(value);});System.out.println("------------------------------------");/** 合并同一商品的数量* */JavaPairRDD<Integer, Product> productRDD2 = productRDD.reduceByKey(new Function2<Product, Product, Product>(){@Overridepublic Product call(Product product, Product product2) throws Exception {product2.quantity += product.quantity;return product2;}});// 按照商品id升序排序JavaPairRDD<Integer, Product> fourproductRankDescRDD = productRDD2.sortByKey(true);System.out.printf("%-10s %-20s %-10s%n", "product_id", "product_name", "total_sales");fourproductRankDescRDD.foreach(tuple -> {Product value = tuple._2;System.out.println(value);});// 将 JavaPairRDD 转换为 JavaRDD<Product>JavaRDD<Product> productRDD3 = productRDD2.values();// 按照 quantity 降序排序JavaRDD<Product> sortedByQuantityRDD = productRDD3.sortBy(product -> product.quantity, false, 1);Scanner scanner = new Scanner(System.in);System.out.print("请输入要显示的前N名商品：");int N = scanner.nextInt();System.out.printf("%-10s %-20s %-10s%n", "product_id", "product_name", "total_sales");sortedByQuantityRDD.take(N).forEach(product -> System.out.println(product));}
}
- IDEA打包:https://blog.csdn.net/kelekele111/article/details/123047189
- 终端运行
/usr/local/spark/bin/spark-submit ~/Desktop/Spark.jar
)
