java - 如何提高 Spark 性能？

Question

我有一个可以处理大型数据集的 Java 程序。数据集存储在 hdfs (csv) 中。

程序运行良好，但速度很慢。

程序做什么:

1. 载入csv文件
2. String[] 单独一行
3. 过滤字符串数组
4. 映射到 MyObject
5. 将 MyObject 保存到 Cassandra

这是我的主要方法:

public static void main(String[] args) {

        // configure spark
        SparkConf sparkConf = new SparkConf().setAppName("Write to cassandra app")
                .setMaster("local[*]")
                .set("spark.executor.memory", "4g");

        if (args.length > 1)
            sparkConf.set("spark.cassandra.connection.host", args[1]);

        // start a spark context
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        // read text file to RDD
        JavaRDD<String> lines = sc.textFile(args[0]);

        JavaRDD<MyObject> myObjectJavaRDD = lines
                .map(line -> line.split(","))
                .filter(someFilter)
                .map(MyObject::new);

        javaFunctions(myObjectJavaRDD).writerBuilder("ks", "table", mapToRow(MyObject.class)).saveToCassandra();
    }

我怎样才能提高性能？

感谢您的回答。

Answer 1

您的代码没有随机播放问题(除非您必须写出到 HDFS)并且默认分区由输入格式定义，在 Hadoop 上由 HDFS 核心分割，过滤器或映射不会更改分区。如果你能先过滤，你会看到一些改进

        JavaRDD<MyObject> myObjectJavaRDD = lines
                .filter(someFilter)
                .map(line -> line.split(","))
                .map(MyObject::new);

Spark 只能为 RDD 的每个分区运行 1 个并发任务，最多为集群中的核心。所以如果你有一个有 50 个核心的集群，你希望你的 RDD 至少有50个分区。至于选择“好”数量的分区，您通常至少需要与并行执行器的数量。您可以通过调用获得此计算值

sc.defaultParallelism

或检查RDD分区数

someRDD.partitions.size

通过使用读取文件创建RDD时

rdd = SparkContext().textFile("hdfs://…/file.txt") 

分区数可能会少一些。理想情况下，你会得到相同的您在 HDFS 中看到的 block 数，但是如果文件中的行太长(长于 block 大小)，将会有更少的分区。

为RDD设置分区数的首选方法是直接将其作为调用中的第二个输入参数

rdd = sc.textFile("hdfs://… /file.txt", 400) 

其中 400 是分区数。在这种情况下，分区会进行 400 次拆分由 Hadoop 的 TextInputFormat 完成，而不是 Spark，它会工作得更快。它的此外，代码会产生 400 个并发任务以尝试将 file.txt 直接加载到 400分区。

重新分区:增加分区，过滤后重新平衡分区增加并行度

        repartition(numPartitions: Int)

合并:在输出到 HDFS/外部之前减少分区而不进行混洗合并

    coalesce(numPartitions: Int, suffle: Boolean = false)

最后，同样重要的是，您可以使用不同的值和基准进行一些试验，以查看该过程花费了多少时间

  val start = System.nanoTime()

  // my process

  val end = System.nanoTime()

  val time = end - start
  println(s"My App takes: $time")

希望对你有帮助