Apache Spark：Java RDD中特定字段的记录计数

Question

我想根据Object中的一个字段统计Java RDD中不同类型的记录。

我有一个实体类，将name 和state 作为类的成员变量。 Entity 类如下所示：

import java.io.Serializable;

import lombok.AllArgsConstructor;
import lombok.Getter;

@Getter
@AllArgsConstructor
public class Entity implements Serializable {
    private final String name;
    private final String state;
}

我有一个实体对象的 javaRDD。我想确定此 RDD 中每个状态存在多少对象。

我目前使用的方法是使用 LongAccumulator。想法是遍历RDD中的每条记录，解析状态字段，并增加相应累加器的计数。到目前为止我尝试过的代码是：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.util.LongAccumulator;

import java.util.ArrayList;
import java.util.List;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class CountRDD {
    public static void main(String[] args) {
        String applicationName = CountRDD.class.getName();
        SparkConf sparkConf = new SparkConf().setAppName(applicationName).setMaster("local");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
        javaSparkContext.setLogLevel("INFO");

        Entity entity1 = new Entity("a1", "s1");
        Entity entity2 = new Entity("a2", "s2");
        Entity entity3 = new Entity("a3", "s1");
        Entity entity4 = new Entity("a4", "s2");
        Entity entity5 = new Entity("a5", "s1");
        List<Entity> entityList = new ArrayList<Entity>();
        entityList.add(entity1);
        entityList.add(entity2);
        entityList.add(entity3);
        entityList.add(entity4);
        entityList.add(entity5);

        JavaRDD<Entity> entityJavaRDD = javaSparkContext.parallelize(entityList, 1);
        LongAccumulator s1Accumulator = javaSparkContext.sc().longAccumulator("s1");
        LongAccumulator s2Accumulator = javaSparkContext.sc().longAccumulator("s2");

        entityJavaRDD.foreach(entity -> {
            if (entity != null) {
                String state = entity.getState();
                if ("s1".equalsIgnoreCase(state)) {
                    s1Accumulator.add(1);
                } else if ("s2".equalsIgnoreCase(state)) {
                    s2Accumulator.add(1);
                }
            }
        });
        log.info("Final values for input entity RDD are following");
        log.info("s1Accumulator = {} ", s1Accumulator.value());
        log.info("s2Accumulator = {} ", s2Accumulator.value());
    }
}

上面的代码工作并产生这个输出s1Accumulator = 3和s2Accumulator = 2。

上述代码的限制是：我们应该在执行前知道状态的所有允许值，并维护相应的累加器。这只会使代码太大而无法获得更大的状态值。

我能想到的另一种方法是创建一个字符串（状态）和整数（计数）的新 Pair RDD。对输入 RDD 应用mapToPair 转换，并从这个新创建的 RDD 中获取计数。

关于如何解决这个问题的任何其他想法。

Answer 1

如 cmets 中所述，您可以在状态字段上groupBy，然后在其上调用count，这将为您提供每个状态的计数。你不需要蓄电池。

附带说明，如果您避免使用 lambda 函数并使用 DataFrames（即DataSet<Row>），作业的运行性能会显着提高。 DataFrames 提供比 RDDs 更好的查询优化和代码生成能力，并为大多数处理用例提供矢量化（意味着：非常快）函数。

DataSet API javadoc 在描述中有一个 DataFrame groupBy 示例：https://spark.apache.org/docs/2.4.5/api/java/org/apache/spark/sql/Dataset.html

首选将数据作为 DataFrame 开始读取，但您可以使用 SparkSession.createDateFrame 转换 RDD 和 JavaRDD。