Skip to content

[pos] Support building presto-on-spark for spark2.4 and spark3.4 #25552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
singcha merged 1 commit into from
Aug 21, 2025
Merged

[pos] Support building presto-on-spark for spark2.4 and spark3.4 #25552

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,8 @@
<dep.guice.version>6.0.0</dep.guice.version>
<dep.arrow.version>17.0.0</dep.arrow.version>

<dep.pos.classloader.module-name.suffix>2</dep.pos.classloader.module-name.suffix>

<!--
America/Bahia_Banderas has:
- offset change since 1970 (offset Jan 1970: -07:00, offset Jan 2018: -06:00)
Expand Down Expand Up @@ -216,6 +218,7 @@
<module>presto-function-server</module>
<module>presto-router-example-plugin-scheduler</module>
<module>presto-plan-checker-router-plugin</module>
<module>presto-spark-classloader-spark${dep.pos.classloader.module-name.suffix}</module>
</modules>

<dependencyManagement>
Expand Down Expand Up @@ -901,6 +904,12 @@
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>com.facebook.presto</groupId>
<artifactId>presto-spark-classloader-spark${dep.pos.classloader.module-name.suffix}</artifactId>
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>com.facebook.presto</groupId>
<artifactId>presto-spark</artifactId>
Expand Down Expand Up @@ -3119,5 +3128,84 @@
</plugins>
</build>
</profile>
<profile>
<id>spark3</id>

<activation>
<property>
<name>spark-version</name>
<value>3</value>
</property>
</activation>

<properties>
<dep.pos.classloader.module-name.suffix>3</dep.pos.classloader.module-name.suffix>
</properties>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.facebook.presto.spark</groupId>
<artifactId>spark-core</artifactId>
<version>3.4.1-1</version>
<scope>provided</scope>
</dependency>
</dependencies>
</dependencyManagement>

<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-enforcer-plugin</artifactId>
<version>3.3.0</version>
<dependencies>
<dependency>
<groupId>org.codehaus.mojo</groupId>
<artifactId>extra-enforcer-rules</artifactId>
<version>1.6.2</version>
</dependency>
</dependencies>
<configuration>
<skip>true</skip>
<rules>
<requireUpperBoundDeps>
<excludes combine.children="append">
<!-- TODO: fix this in Airlift resolver -->
<exclude>org.codehaus.plexus:plexus-utils</exclude>
<exclude>com.google.guava:guava</exclude>
<exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
<exclude>com.fasterxml.jackson.core:jackson-core</exclude>
<exclude>com.fasterxml.jackson.core:jackson-databind</exclude>
</excludes>
</requireUpperBoundDeps>
</rules>
</configuration>
</plugin>

<plugin>
<groupId>org.basepom.maven</groupId>
<artifactId>duplicate-finder-maven-plugin</artifactId>
<configuration>
<skip>true</skip>
<ignoredClassPatterns combine.children="append">
<ignoredClassPattern>com.github.benmanes.caffeine.*</ignoredClassPattern>
<!-- Duplicate class is being brought in by commons-io & log4j-api -->
<ignoredClassPattern>META-INF.versions.9.module-info</ignoredClassPattern>
<!-- Duplicate class is being brought in by several netty dependencies-->
<ignoredClassPattern>META-INF.versions.11.module-info</ignoredClassPattern>
<!-- Ignore duplicate classes related to lucene-core and ranger-apache -->
<ignoredClassPattern>META-INF.versions.9.org.apache.lucene.*</ignoredClassPattern>
</ignoredClassPatterns>

</configuration>
</plugin>

</plugins>
</pluginManagement>
</build>

</profile>
</profiles>
</project>
113 changes: 112 additions & 1 deletion presto-spark-base/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
<dep.jetty.version>9.4.55.v20240627</dep.jetty.version>
<dep.okhttp.version>4.12.0</dep.okhttp.version>
<dep.okio-jvm.version>3.9.1</dep.okio-jvm.version>
<dep.pos.classloader.module-name.suffix>2</dep.pos.classloader.module-name.suffix>
</properties>

<dependencyManagement>
Expand Down Expand Up @@ -51,7 +52,12 @@
<dependency>
<groupId>com.facebook.presto</groupId>
<artifactId>presto-spark-classloader-interface</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>com.facebook.presto</groupId>
<artifactId>presto-spark-classloader-spark${dep.pos.classloader.module-name.suffix}</artifactId>
<scope>provided</scope>
</dependency>

Expand Down Expand Up @@ -459,6 +465,10 @@
<ignoredUnusedDeclaredDependencies>
<ignoredUnusedDeclaredDependency>javax.ws.rs:javax.ws.rs-api</ignoredUnusedDeclaredDependency>
<ignoredUnusedDeclaredDependency>javax.servlet:javax.servlet-api</ignoredUnusedDeclaredDependency>
<ignoredUnusedDeclaredDependency>com.google.code.findbugs:jsr305</ignoredUnusedDeclaredDependency>
<ignoredUnusedDeclaredDependency>javax.inject:javax.inject</ignoredUnusedDeclaredDependency>
<ignoredUnusedDeclaredDependency>javax.annotation:javax.annotation-api</ignoredUnusedDeclaredDependency>
<ignoredUnusedDeclaredDependency>org.apache.httpcomponents:httpcore</ignoredUnusedDeclaredDependency>
</ignoredUnusedDeclaredDependencies>
</configuration>
</plugin>
Expand Down Expand Up @@ -524,5 +534,106 @@
</plugins>
</build>
</profile>
<profile>
<id>spark3</id>

<activation>
<property>
<name>spark-version</name>
<value>3</value>
</property>
</activation>

<properties>
<dep.pos.classloader.module-name.suffix>3</dep.pos.classloader.module-name.suffix>
</properties>

<dependencies>
<dependency>
<groupId>com.facebook.presto.spark</groupId>
<artifactId>spark-core</artifactId>
<version>3.4.1-1</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>com.google.inject</groupId>
<artifactId>guice</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>jakarta.annotation</groupId>
<artifactId>jakarta.annotation-api</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>com.google.errorprone</groupId>
<artifactId>error_prone_annotations</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.weakref</groupId>
<artifactId>jmxutils</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>jakarta.validation</groupId>
<artifactId>jakarta.validation-api</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.13.8</version>
<scope>provided</scope>
</dependency>

</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<configuration>
<ignoredDependencies>
<ignoredDependency>org.glassfish.hk2.external:jakarta.inject</ignoredDependency>
<ignoredDependency>org.apache.hadoop:hadoop-client-api</ignoredDependency>
<ignoredDependency>jakarta.annotation:jakarta.annotation-api</ignoredDependency>
<ignoredDependency>jakarta.validation:jakarta.validation-api</ignoredDependency>
</ignoredDependencies>
</configuration>
</plugin>

<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<testExcludes>
<!-- skip TestPrestoSparkExecutionExceptionFactory UT to avoid errors of incompatible types between spark 2 and spark 3-->
<exclude>**/TestPrestoSparkExecutionExceptionFactory.java</exclude>
</testExcludes>
</configuration>
</plugin>

<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<!-- skip TestPrestoSparkExecutionExceptionFactory UT to avoid errors of incompatible types between spark 2 and spark 3-->
<exclude>**/TestPrestoSparkExecutionExceptionFactory.java</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>

</profile>
</profiles>
</project>
46 changes: 46 additions & 0 deletions presto-spark-classloader-interface/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
<properties>
<air.main.basedir>${project.parent.basedir}</air.main.basedir>
<air.check.skip-modernizer>true</air.check.skip-modernizer>
<dep.pos.classloader.module-name.suffix>2</dep.pos.classloader.module-name.suffix>
</properties>

<dependencies>
Expand All @@ -21,14 +22,59 @@
<artifactId>spark-core</artifactId>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>com.facebook.presto</groupId>
<artifactId>presto-spark-classloader-spark${dep.pos.classloader.module-name.suffix}</artifactId>
</dependency>

<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>

<dependency>
<groupId>com.facebook.airlift</groupId>
<artifactId>units</artifactId>
</dependency>
</dependencies>

<profiles>
<profile>
<id>spark3</id>

<activation>
<property>
<name>spark-version</name>
<value>3</value>
</property>
</activation>

<properties>
<dep.pos.classloader.module-name.suffix>3</dep.pos.classloader.module-name.suffix>
</properties>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.facebook.presto.spark</groupId>
<artifactId>spark-core</artifactId>
<version>3.4.1-1</version>
<scope>compile</scope>
</dependency>
</dependencies>
</dependencyManagement>

<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.13.8</version>
<scope>provided</scope>
</dependency>
</dependencies>

</profile>
</profiles>

</project>
22 changes: 12 additions & 10 deletions ...c/main/java/com/facebook/presto/spark/classloader_interface/PrestoSparkNativeTaskRdd.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.spark.storage.BlockId;
import org.apache.spark.storage.BlockManagerId;
import scala.Tuple2;
import scala.Tuple3;
import scala.collection.Iterator;
import scala.collection.Seq;

Expand All @@ -38,12 +39,13 @@
import java.util.Optional;
import java.util.stream.Collectors;

import static com.facebook.presto.spark.classloader_interface.PrestoSparkUtils.asJavaCollection;
import static com.facebook.presto.spark.classloader_interface.PrestoSparkUtils.getMapSizesByExecutorId;
import static com.facebook.presto.spark.classloader_interface.PrestoSparkUtils.seqAsJavaList;
import static com.facebook.presto.spark.classloader_interface.ScalaUtils.emptyScalaIterator;
import static com.google.common.base.Preconditions.checkState;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static scala.collection.JavaConversions.asJavaCollection;
import static scala.collection.JavaConversions.seqAsJavaList;

/**
* PrestoSparkTaskRdd represents execution of Presto stage, it contains:
Expand Down Expand Up @@ -173,34 +175,34 @@ private Optional<PrestoSparkShuffleWriteDescriptor> getShuffleWriteDescriptor(in
private List<String> getBlockIds(ShuffledRDDPartition partition, ShuffleHandle shuffleHandle)
{
MapOutputTracker mapOutputTracker = SparkEnv.get().mapOutputTracker();
Collection<Tuple2<BlockManagerId, Seq<Tuple2<BlockId, Object>>>> mapSizes = asJavaCollection(mapOutputTracker.getMapSizesByExecutorId(
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1));
Collection<Tuple2<BlockManagerId, Seq<Tuple3<BlockId, Object, Object>>>> mapSizes = getMapSizesByExecutorId(mapOutputTracker,
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1);
return mapSizes.stream().map(item -> item._1.executorId()).collect(Collectors.toList());
}

private List<String> getPartitionIds(ShuffledRDDPartition partition, ShuffleHandle shuffleHandle)
{
MapOutputTracker mapOutputTracker = SparkEnv.get().mapOutputTracker();
Collection<Tuple2<BlockManagerId, Seq<Tuple2<BlockId, Object>>>> mapSizes = asJavaCollection(mapOutputTracker.getMapSizesByExecutorId(
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1));
Collection<Tuple2<BlockManagerId, Seq<Tuple3<BlockId, Object, Object>>>> mapSizes = getMapSizesByExecutorId(mapOutputTracker,
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1);
return mapSizes.stream()
.map(item -> asJavaCollection(item._2))
.flatMap(Collection::stream)
.map(i -> i._1.toString())
.map(i -> i._1().toString())
.collect(Collectors.toList());
}

private List<Long> getPartitionSize(ShuffledRDDPartition partition, ShuffleHandle shuffleHandle)
{
MapOutputTracker mapOutputTracker = SparkEnv.get().mapOutputTracker();
Collection<Tuple2<BlockManagerId, Seq<Tuple2<BlockId, Object>>>> mapSizes = asJavaCollection(mapOutputTracker.getMapSizesByExecutorId(
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1));
Collection<Tuple2<BlockManagerId, Seq<Tuple3<BlockId, Object, Object>>>> mapSizes = getMapSizesByExecutorId(mapOutputTracker,
shuffleHandle.shuffleId(), partition.idx(), partition.idx() + 1);
//Each partition/BlockManagerId can contain multiple blocks (with BlockId), here sums up all the blocks from each BlockManagerId/Partition
return mapSizes.stream()
.map(
item -> seqAsJavaList(item._2)
.stream()
.map(item2 -> ((Long) item2._2))
.map(item2 -> ((Long) item2._2()))
.reduce(0L, Long::sum))
.collect(Collectors.toList());
}
Expand Down
Loading
Loading