diff --git a/docs/usage/properties/instance/user/bulk_import.md b/docs/usage/properties/instance/user/bulk_import.md index 98c7d4ea788..f089d4f992e 100644 --- a/docs/usage/properties/instance/user/bulk_import.md +++ b/docs/usage/properties/instance/user/bulk_import.md @@ -109,8 +109,8 @@ Note that on EMR, the total resource allocation must align with the instance typ | sleeper.bulk.import.eks.spark.default.parallelism | (EKS mode only) The default parallelism for the Spark job. Used to set spark.default.parallelism. Should scale with the total cores across the EKS cluster, which may differ from EMR.
See https://spark.apache.org/docs/latest/configuration.html. | 290 | false | | sleeper.bulk.import.eks.spark.sql.shuffle.partitions | (EKS mode only) The number of partitions used in a Spark SQL/dataframe shuffle operation. Used to set spark.sql.shuffle.partitions.
See https://spark.apache.org/docs/latest/configuration.html. | 290 | false | | sleeper.bulk.import.eks.spark.dynamic.allocation.enabled | (EKS mode only) Whether Spark should use dynamic allocation to scale resources up and down. Used to set spark.dynamicAllocation.enabled. Kubernetes support for dynamic allocation is more limited than YARN's; consider leaving this disabled on EKS.
See https://spark.apache.org/docs/latest/configuration.html. | false | false | -| sleeper.bulk.import.eks.spark.executor.extra.java.options | JVM options passed to the executors. Used to set spark.executor.extraJavaOptions.
See https://spark.apache.org/docs/latest/configuration.html. | -XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' | false | -| sleeper.bulk.import.eks.spark.driver.extra.java.options | JVM options passed to the driver. Used to set spark.driver.extraJavaOptions.
See https://spark.apache.org/docs/latest/configuration.html. | -XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' | false | +| sleeper.bulk.import.eks.spark.executor.extra.java.options | JVM options passed to the executors. Used to set spark.executor.extraJavaOptions.
See https://spark.apache.org/docs/latest/configuration.html. | -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' | false | +| sleeper.bulk.import.eks.spark.driver.extra.java.options | JVM options passed to the driver. Used to set spark.driver.extraJavaOptions.
See https://spark.apache.org/docs/latest/configuration.html. | -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' | false | | sleeper.bulk.import.eks.spark.executor.heartbeat.interval | The interval between heartbeats from executors to the driver. Used to set spark.executor.heartbeatInterval.
See https://spark.apache.org/docs/latest/configuration.html. | 60s | false | | sleeper.bulk.import.eks.spark.network.timeout | The default timeout for network interactions in Spark. Used to set spark.network.timeout.
See https://spark.apache.org/docs/latest/configuration.html. | 800s | false | | sleeper.bulk.import.eks.spark.memory.fraction | The fraction of heap space used for execution and storage. Used to set spark.memory.fraction.
See https://spark.apache.org/docs/latest/configuration.html. | 0.80 | false | diff --git a/example/full/instance.properties b/example/full/instance.properties index 84480d7d91e..e8990c196f3 100644 --- a/example/full/instance.properties +++ b/example/full/instance.properties @@ -1247,12 +1247,12 @@ # JVM options passed to the executors. Used to set spark.executor.extraJavaOptions. # See https://spark.apache.org/docs/latest/configuration.html. # (default value shown below, uncomment to set a value) -# sleeper.bulk.import.eks.spark.executor.extra.java.options=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' +# sleeper.bulk.import.eks.spark.executor.extra.java.options=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' # JVM options passed to the driver. Used to set spark.driver.extraJavaOptions. # See https://spark.apache.org/docs/latest/configuration.html. # (default value shown below, uncomment to set a value) -# sleeper.bulk.import.eks.spark.driver.extra.java.options=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' +# sleeper.bulk.import.eks.spark.driver.extra.java.options=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' # The interval between heartbeats from executors to the driver. Used to set # spark.executor.heartbeatInterval. diff --git a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/configuration/SparkConfigurationUtils.java b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/configuration/SparkConfigurationUtils.java index 49ddb27c87e..b213f8e5690 100644 --- a/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/configuration/SparkConfigurationUtils.java +++ b/java/bulk-import/bulk-import-core/src/main/java/sleeper/bulkimport/core/configuration/SparkConfigurationUtils.java @@ -214,6 +214,7 @@ public static Map getSparkConfigurationForEKSFromInstancePropert sparkConf.put("spark.kubernetes.namespace", instanceProperties.get(BULK_IMPORT_EKS_NAMESPACE)); sparkConf.put("spark.kubernetes.authenticate.driver.serviceAccountName", "spark"); sparkConf.put("spark.kubernetes.executor.podTemplateFile", "/tmp/executor-template.yaml"); + sparkConf.put("spark.kubernetes.executor.podTemplateContainerName", "spark-kubernetes-executor"); // Ensure pods aren't disrupted/evicted before they finish sparkConf.put("spark.kubernetes.driver.annotation.karpenter.sh/do-not-disrupt", "true"); diff --git a/java/bulk-import/bulk-import-eks/docker/eks/Dockerfile b/java/bulk-import/bulk-import-eks/docker/eks/Dockerfile index 42886952c5b..ea5d2043edf 100644 --- a/java/bulk-import/bulk-import-eks/docker/eks/Dockerfile +++ b/java/bulk-import/bulk-import-eks/docker/eks/Dockerfile @@ -12,23 +12,16 @@ # See the License for the specific language governing permissions and # limitations under the License. -ARG BASE_IMAGE=apache/spark:3.5.6-scala2.12-java17-ubuntu +ARG BASE_IMAGE=public.ecr.aws/emr-on-eks/spark/emr-7.12.0:latest FROM ${BASE_IMAGE} -ENV PATH="$PATH:/opt/spark/bin" USER root -RUN apt-get update && apt-get upgrade -y && rm -rf /var/lib/apt/lists/* -RUN rm /opt/spark/jars/* -RUN mkdir /opt/spark/workdir -USER spark -# Replace Spark jars with versions managed by Sleeper (the version of Spark must match) -COPY ./spark/* /opt/spark/jars -COPY ./bulk-import-runner.jar /opt/spark/workdir +RUN mkdir -p /opt/spark/workdir && chown hadoop:hadoop /opt/spark/workdir +COPY --chown=hadoop:hadoop ./bulk-import-runner.jar /opt/spark/workdir/ -USER root COPY ./sleeper-entrypoint.sh /opt/sleeper-entrypoint.sh RUN chmod +x /opt/sleeper-entrypoint.sh -USER spark +USER hadoop:hadoop ENTRYPOINT ["/opt/sleeper-entrypoint.sh"] diff --git a/java/bulk-import/bulk-import-eks/docker/eks/sleeper-entrypoint.sh b/java/bulk-import/bulk-import-eks/docker/eks/sleeper-entrypoint.sh index ea80772e36c..edacee99945 100755 --- a/java/bulk-import/bulk-import-eks/docker/eks/sleeper-entrypoint.sh +++ b/java/bulk-import/bulk-import-eks/docker/eks/sleeper-entrypoint.sh @@ -19,4 +19,4 @@ if [ -n "${EXECUTOR_POD_TEMPLATE:-}" ]; then printf '%s' "$EXECUTOR_POD_TEMPLATE" > /tmp/executor-template.yaml fi -exec /opt/entrypoint.sh "$@" +exec /usr/bin/entrypoint.sh "$@" diff --git a/java/bulk-import/bulk-import-eks/pom.xml b/java/bulk-import/bulk-import-eks/pom.xml index dc619ebd259..63f0c5fbec7 100644 --- a/java/bulk-import/bulk-import-eks/pom.xml +++ b/java/bulk-import/bulk-import-eks/pom.xml @@ -24,46 +24,4 @@ 4.0.0 bulk-import-eks - - - - org.apache.spark - spark-sql_${scala.version} - runtime - - - org.apache.spark - spark-kubernetes_${scala.version} - runtime - - - org.apache.arrow - arrow-memory-unsafe - runtime - - - - - - - maven-dependency-plugin - - - - - - - - - copy-dependencies - - - ${project.build.directory}/spark - runtime - - - - - - \ No newline at end of file diff --git a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/StateMachinePlatformExecutor.java b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/StateMachinePlatformExecutor.java index 8b2c78a93fd..63747bf2884 100644 --- a/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/StateMachinePlatformExecutor.java +++ b/java/bulk-import/bulk-import-starter/src/main/java/sleeper/bulkimport/starter/executor/StateMachinePlatformExecutor.java @@ -37,7 +37,6 @@ */ public class StateMachinePlatformExecutor implements PlatformExecutor { private static final String SPARK_IMAGE_JAR_LOCATION = "local:///opt/spark/workdir/bulk-import-runner.jar"; - private static final String SPARK_IMAGE_JAVA_HOME = "/opt/java/openjdk"; private static final String NATIVE_IMAGE_JAR_LOCATION = "local:///opt/spark/workdir/bulk-import-runner.jar"; private static final String NATIVE_IMAGE_LOG4J_LOCATION = "file:///opt/spark/workdir/log4j.properties"; private static final String NATIVE_IMAGE_JAVA_HOME = "/usr/lib/jvm/java-11-amazon-corretto"; @@ -92,7 +91,6 @@ private List constructArgs(BulkImportArguments arguments, String taskId) baseSparkConfig.put("spark.executor.extraJavaOptions", "-Dlog4j.configuration=" + NATIVE_IMAGE_LOG4J_LOCATION); jarLocation = NATIVE_IMAGE_JAR_LOCATION; } else { - baseSparkConfig.put("spark.executorEnv.JAVA_HOME", SPARK_IMAGE_JAVA_HOME); jarLocation = SPARK_IMAGE_JAR_LOCATION; } diff --git a/java/bulk-import/bulk-import-starter/src/test/resources/example/step-functions/startexecution-input.json b/java/bulk-import/bulk-import-starter/src/test/resources/example/step-functions/startexecution-input.json index 9b6a7045251..55c278bcfb6 100644 --- a/java/bulk-import/bulk-import-starter/src/test/resources/example/step-functions/startexecution-input.json +++ b/java/bulk-import/bulk-import-starter/src/test/resources/example/step-functions/startexecution-input.json @@ -12,7 +12,7 @@ "--conf", "spark.driver.cores=5", "--conf", - "spark.driver.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'", + "spark.driver.extraJavaOptions=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'", "--conf", "spark.driver.memory=12g", "--conf", @@ -22,7 +22,7 @@ "--conf", "spark.executor.cores=5", "--conf", - "spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'", + "spark.executor.extraJavaOptions=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'", "--conf", "spark.executor.heartbeatInterval=60s", "--conf", @@ -32,8 +32,6 @@ "--conf", "spark.executor.memoryOverhead=1g", "--conf", - "spark.executorEnv.JAVA_HOME=/opt/java/openjdk", - "--conf", "spark.hadoop.fs.s3a.aws.credentials.provider=software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider", "--conf", "spark.hadoop.fs.s3a.block.size=32M", @@ -62,6 +60,8 @@ "--conf", "spark.kubernetes.executor.podNamePrefix=job-test-job", "--conf", + "spark.kubernetes.executor.podTemplateContainerName=spark-kubernetes-executor", + "--conf", "spark.kubernetes.executor.podTemplateFile=/tmp/executor-template.yaml", "--conf", "spark.kubernetes.namespace=eks-namespace", diff --git a/java/core/src/main/java/sleeper/core/properties/instance/EKSProperty.java b/java/core/src/main/java/sleeper/core/properties/instance/EKSProperty.java index 2cea4968489..b9790270566 100644 --- a/java/core/src/main/java/sleeper/core/properties/instance/EKSProperty.java +++ b/java/core/src/main/java/sleeper/core/properties/instance/EKSProperty.java @@ -144,7 +144,7 @@ public interface EKSProperty { .description("JVM options passed to the executors. Used to set spark.executor.extraJavaOptions.\n" + "See https://spark.apache.org/docs/latest/configuration.html.") .defaultValue( - "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'") + "-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'") .propertyGroup(InstancePropertyGroup.BULK_IMPORT).build(); UserDefinedInstanceProperty BULK_IMPORT_EKS_SPARK_DRIVER_EXTRA_JAVA_OPTIONS = Index.propertyBuilder("sleeper.bulk.import.eks.spark.driver.extra.java.options") .description("JVM options passed to the driver. Used to set spark.driver.extraJavaOptions.\n" + diff --git a/scripts/templates/instanceproperties.template b/scripts/templates/instanceproperties.template index d8bc4798ad2..32a29379ead 100644 --- a/scripts/templates/instanceproperties.template +++ b/scripts/templates/instanceproperties.template @@ -1251,12 +1251,12 @@ # JVM options passed to the executors. Used to set spark.executor.extraJavaOptions. # See https://spark.apache.org/docs/latest/configuration.html. # (default value shown below, uncomment to set a value) -# sleeper.bulk.import.eks.spark.executor.extra.java.options=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' +# sleeper.bulk.import.eks.spark.executor.extra.java.options=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' # JVM options passed to the driver. Used to set spark.driver.extraJavaOptions. # See https://spark.apache.org/docs/latest/configuration.html. # (default value shown below, uncomment to set a value) -# sleeper.bulk.import.eks.spark.driver.extra.java.options=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' +# sleeper.bulk.import.eks.spark.driver.extra.java.options=-XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' # The interval between heartbeats from executors to the driver. Used to set # spark.executor.heartbeatInterval.