Java spark groupByKey 和 key1 并在 groupedRDD 上使用 key2 做聚合

Question

我正在尝试做一个简单的 java spark 应用程序，它执行以下操作

输入数据csv格式：key1,key2,data1,data2

基本上我在这里要做的是，

首先我通过 key1 映射每一行，然后对该 rdd 执行 groupByKey 操作。

JavaRDD<String> viewRdd = sc.textFile("testfile.csv", 1);
JavaPairRDD<String, String> customerIdToRecordRDD = viewRdd
    .mapToPair(w -> new Tuple2<String, String>(w.split(",")[0], w));
JavaPairRDD<String, Iterable<String>> groupedByKey1RDD = customerIdToRecordRDD.groupByKey();
System.out.println(customerIdToRecordGropedRDD.count());

现在我的问题是，我需要在 groupedByKey1RDD 的每个组上使用 key2 进行聚合。有什么方法可以将 Iterable 转换为 RDD 吗？还是我在这里遗漏了什么。我是新手，任何帮助都会

示例输入和预期输出：

id_1,time0,10,10

id_2,time1,0,10

id_1,time1,11,10

id_1,time0,1,10

id_2,time1,10,10

输出按第1列分组，然后按第2列聚合（聚合逻辑是简单地将column3和column相加）：

id_1 : time0 : { sum1 : 11, sum2 : 20} ,
       time1 : { sum1 : 11, sum2 : 10}

id_2 : time1 : { sum1 : 10, sum2 : 20} 

Answer 1

这是使用 Spark 2.0 和 Dataframe 的解决方案。如果您仍想使用 RDD，请告诉我。

public class SparkGroupBySample {
    public static void main(String[] args) {
    //SparkSession
    SparkSession spark = SparkSession
            .builder()
            .appName("SparkGroupBySample")
            .master("local")
            .getOrCreate();     
    //Schema
    StructType schema = new StructType(new StructField[] { 
            new StructField("key1", DataTypes.StringType, true, Metadata.empty()),
            new StructField("key2", DataTypes.StringType, true, Metadata.empty()),
            new StructField("data1", DataTypes.IntegerType, true, Metadata.empty()),
            new StructField("data2", DataTypes.IntegerType, true, Metadata.empty())});
    //Read csv
    Dataset<Row> dataSet = spark.read().format("csv").schema(schema).option("header", "true").option("delimiter", ",").load("c:\\temp\\sample.csv");
    dataSet.show();     
    //groupBy and aggregate
    Dataset<Row> dataSet1 = dataSet.groupBy("key1","key2").sum("data1","data2").toDF("key1","key2","sum1","sum2");
    dataSet1.show();
    //stop
    spark.stop();
   }
}

这是输出。

+----+-----+----+----+
|key1| key2|sum1|sum2|
+----+-----+----+----+
|id_1|time1|  11|  10|
|id_2|time1|  10|  20|
|id_1|time0|  11|  20|
+----+-----+----+----+