如何改进我Spark应用程序中的reducebykey部分？

Question

13 浏览2023年3月30日

匿名的 2023年3月31日

0 Comments

我有64个Spark核心。我有超过8000万行的数据，占据了我的Cassandra集群的4.2 GB空间。现在我需要82秒来处理这些数据。我希望将这个时间缩短到8秒。你对此有什么想法吗？这个目标可行吗？谢谢。

这是我想改进的Spark应用的部分代码：

axes = sqlContext.read.format("org.apache.spark.sql.cassandra")\
    .options(table="axes", keyspace=source, numPartitions="192").load()\
    .repartition(64*3)\
    .reduceByKey(lambda x,y:x+y,52)\
    .map(lambda x:(x.article,[Row(article=x.article,at=x.at,comments=x.comments,likes=x.likes,reads=x.reads,shares=x.shares)]))\
    .map(lambda x:(x[0],sorted(x[1],key=lambda y:y.at,reverse = False))) \
    .filter(lambda x:len(x[1])>=2) \
    .map(lambda x:x[1][-1])

编辑：

这是我目前运行的代码，上面发布的代码只是一个实验，对于造成的混淆抱歉。上述问题与此代码有关。

axes = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="axes", keyspace=source).load().repartition(64*3) \
                    .map(lambda x:(x.article,[Row(article=x.article,at=x.at,comments=x.comments,likes=x.likes,reads=x.reads,shares=x.shares)])).reduceByKey(lambda x,y:x+y)\
                    .map(lambda x:(x[0],sorted(x[1],key=lambda y:y.at,reverse = False))) \
                    .filter(lambda x:len(x[1])>=2) \
                    .map(lambda x:x[1][-1])

谢谢。

0

1 答案

匿名的 · Answer 1 · 2023-05-19T09:21:23+00:00

问题：如何改进我的Spark应用程序中的reduceByKey部分？

原因：下面的代码存在一些问题，导致reduceByKey部分无法正确工作。

解决方法：

1. 第一步中的代码创建了一个Spark DataFrame，但是numPartitions选项可能不被识别，需要注意。

2. 第二步中的代码没有实际工作，只是将数据重新分区，这样做不会有任何作用。

3. 在第三步中，代码切换到RDD，但是由于Row是tuple的子类，而reduceByKey只能用于成对的RDD，所以每个元素必须是大小为2的tuple。此外，选择52个分区的原因不明。

4. 在第四步中，由于reduceByKey始终返回大小为2的tuple的RDD，所以以下部分代码将不起作用。特别是，x不能具有像article或comments这样的属性。此外，创建大小为1的列表没有必要。如果存在一些过时的列，应该在将数据转换为RDD之前过滤掉这些列，以避免过多的流量和减少内存使用。如果没有过时的列，则没有理由通过创建新对象增加对Python GC的压力。

5. 在第五步中，由于x[1]只有一个元素，对其进行排序没有意义。

6. 在第六步中，该过滤器将始终返回一个空的RDD。

7. 在第七步中，代码没有执行任何有用的操作。

如果使用了这段代码的某个版本，那么问题很可能是问题中显示的顺序混乱，从第四点开始的map代码：

.map(lambda x: (x.article,[Row(....)]))

在reduceByKey之前：

.reduceByKey(lambda x,y:x+y,52)

如果是这种情况，实际上使用.reduceByKey来执行groupByKey，这要么等效于具有所有问题的groupByKey（Python），要么效率较低（Scala）。此外，减少分区数量的操作也是非常可疑的。如果是这种情况，没有充分的理由将数据移出JVM（DataFrame -> RDD转换），并进行相应的序列化和反序列化，即使有，也可以通过使用实际的最大值进行实际的减少来轻松解决。