Hadoop S3 Configuration Guide

Introduction

This document outlines best practices for tuning S3A client parameters (the Hadoop S3 connector) when using VAST S3. It covers common configuration options, explains the impact of each, and offers guidelines for optimal performance based on typical use cases.

Important Considerations

1. No One-Size-Fits-All Configuration
   Every environment is different. The optimal settings can vary based on workload characteristics, data sizes, network constraints, and the overall cluster configuration.

2. Holistic Performance Evaluation
   Tuning parameters that benefit one workload might negatively affect others. Performance benchmarks should be jointly defined and continuously measured, considering the broader context (e.g., concurrency, resource availability, overall S3 load).

3. Resource Allocation
   High-load scenarios may require more CPU, memory, and network bandwidth. Conversely, smaller workloads or test environments can run effectively with lower settings.

4. Incremental Changes
   It’s often wise to adjust parameters in small increments, validate through tests or pilot runs, and monitor the impact on overall system health and performance.

The configurations outlined below reflect experiences across multiple customer deployments and in-house benchmarks. Adjust them as necessary for your specific use case.

Configurations

S3 Endpoint

When connecting to VAST S3, you have several endpoint options (option fs.s3a.endpoint)

• VIP Endpoint: Points directly to a single CNode via a Virtual IP (VIP). This can be throughput-limited by that one CNode.

• DNS Endpoint: A DNS-based endpoint may distribute connections across multiple CNodes, offering potential load-balancing benefits. Consult VAST support for the correct DNS endpoint.

• Proxy Endpoint (e.g., HAProxy): Allows for custom routing or advanced load-balancing, but adds another network hop.

In your environment, test each endpoint type (VIP, DNS, Proxy) with representative workloads to see which yields the best performance and reliability.

S3a Connection Settings

Recommendation

• Increase these values carefully in conjunction with performance testing.

• Monitor cluster-wide connection usage. High concurrency can impact both your cluster (memory, CPU) and your VAST cluster resources.

Committers

These “committers” define how output data is staged and finalized (committed) when writing to S3

• FileOutputCommitter (Classic or “v1”)

  • The default committer in many Hadoop versions.

  • Stages all output in a temporary directory (e.g. _temporary/0/) and then moves (copies) data to the final destination after all tasks are complete.

  • Can suffer from slow commit times on object stores due to a lack of a true “rename.”

• DirectoryOutputCommitter (Sometimes called “v2”)

  • Allows each task to commit its output immediately upon completion (rather than waiting for the entire job to finish).

  • Can reduce the driver’s commit overhead but makes it more complex to handle failures (partial data may already be in the final location).

• PartitionedOutputCommitter

  • Similar to the directory committer approach but optimized specifically for partitioned data (common in Hive-style tables).

  • Helps reduce the overhead of moving or listing large numbers of partitioned files.

• Magic Committer

  • Writes data directly into “magic” paths (special S3 paths with markers) that trigger atomic rename-like behavior within S3.

  • Eliminates the extra copy step by instructing S3 to finalize the file in place.

  • Requires the job to write to specially formatted “magic” paths (usually s3a://bucket/magic/...)

Each committer faces trade-offs among performance, consistency, and fault tolerance. The best choice often depends on your workload requirements, the size of your data, and how tolerant you are of partial writes or longer commit times.

VAST Hard Link Optimization

Spark stores intermediate data in a staging area to prevent data corruption. During this process, data is initially written to the staging directory and, upon completion, is moved to the target directory. In the case of S3, this operation uses the s3 cp command instead of a traditional mv or rename command, as S3 does not support native rename or move operations.

Due to this limitation, writing to S3a can be slower. To address this performance bottleneck, an optimization called a hard link is introduced. Instead of performing a full copy (cp), a hard link is created, significantly reducing the overhead. Hard link is vast feature providing optimization to commit files (renaming) by not copying them (works only for user-agent Hadoop), As a result, only minimal latency is incurred during the transition from the staging area to the destination directory.

ℹ️ Info

How to apply?

vtool vsettings set S3_COPY_USING_LINK=true - will activate the hard-link flow

Here is an example of the speedup and settings for both MPU and vast S3_COPY_USING_LINK

ℹ️ Info

With MPU enabled and S3_COPY_USING_LINK=true :

$ time aws --profile var204-sh s3 cp s3://shb1/1g s3://shb1/1g-2
copy: s3://shb1/1g to s3://shb1/1g-2

real    0m3.039s
user    0m0.677s
sys     0m0.082s

With MPU disabled and S3_COPY_USING_LINK=true :

$ time aws --profile var204-sh-nompu s3 cp s3://shb1/1g s3://shb1/1g-2
copy: s3://shb1/1g to s3://shb1/1g-2

real    0m0.709s
user    0m0.534s
sys     0m0.063s

Using the directory v2 directory committer

In the default commit protocol, the driver waits for all tasks to complete before initiating the commit process. During task execution, output files are typically written to a staging directory such as s3a://your/destination/folder/_temporary/0/. Once all tasks have finished, the driver performs a commit operation, which involves renaming or moving files from the staging directory to the final destination. This sequential renaming can be problematic; if a failure occurs during this process, some files may reside in the destination directory while others remain in the staging area, leading to inconsistencies.

⚠️

In contrast, the v2 directory committer enhances this process by allowing each task to commit its output immediately upon completion. With this approach, the staging directory is typically s3a://your/destination/folder/_temporary/0/_temporary/. As each task finishes, it performs its own commit operation, moving or renaming its output files to the final destination. While this reduces the time the driver spends waiting for all tasks to finish before committing, it introduces potential risks. If some tasks or the entire job fail, it's possible to find a mix of committed files in both the destination and staging directories, leading to partial or inconsistent data visibility.

It's important to note that while the v2 directory committer can improve performance by overlapping task completion and committing, it requires careful handling to ensure data consistency, especially in failure scenarios.

Monitor the commit time

Spark provides a metric for commit time.

In Spark UI, go to SQL / Dataframe → your stage of writing → Execute InsertIntoHadoopFsRelationCommand → job commit time

Multipart Upload

Multipart splits the target file into equal parts (with the last part being the resilient one).

E.g. - 1GB file with a 32 MB part Size will be split into 33 parts of 32 MB and 1 part of 8 MB (total 34 parts).

70GB - will be split into 2187 parts of 32MB and 1 part of 16MB (total 2188 parts).

Multipart with hard link

When multipart uploads are used, the object is created via a multi-part assembly, not via a final COPY operation. Therefore, the “hard link” optimization only applies to rename-style operations. Still, enabling S3_COPY_USING_LINK is generally beneficial for all other writes and merges within Spark.

Multipart with directory committer

If you enable the v2 directory committer, each task might finalize its parts independently. The underlying assembly is still the same; the key difference is the timing of the commit.

Multipart benchmark

For reference, an internal test uploading 70 GB in 10 iterations showed the best throughput at a 32 MB part size. Smaller or larger part sizes resulted in either too many parts or insufficient parallelism.

Monitor multipart

Spark UI and Logs

• Spark UI: Inspect job commit time, shuffle read/write times, and stage tasks to identify whether S3 operations are a bottleneck.

• Executors’ Logs: Look for HTTP 429 (Too Many Requests) or other errors indicating that you may be pushing the environment too hard.

VAST UI (Analytics → Custom Analytics → S3Metric_multi)

You should have support access (log in with a support user)

• multi_part_upload Number of multipart sessions.

• multi_part_upload_fallback Indicates reallocation events. This can happen when the last part arrives first (especially if the last part is smaller). VAST has internal workarounds for this edge case, and additional improvements may be available in newer versions.

• Throughput and Error Metrics: Check for spikes in concurrency, memory usage, or error rates.

• multi_part_upload_fallback - when we need to re-allocate (In case where the last part arrives first (last part is the smallest part (because it is splited to equal parts so the last one is the left-over), and vast allocate memory for all part based on the first part arrives) - this is unwanted situation - we have workaround in vast where when the first part arrive is bigger than 10 (configurable) we wait few more ms (configurable) for other part - this issue will be solved in newer versions)

• multi_part_upload - number of multi-part sessions

Putting It All Together. A Sample Tuning Checklist

• Establish Baseline

  • Use default S3A and Spark settings.

  • Run a representative workload (e.g., a subset of your typical data processing).

• Enable VAST Hard Link

  • vtool vsettings set S3_COPY_USING_LINK=true

  • Confirm in the logs that hard linking is occurring.

• Adjust Connection Settings

  • fs.s3a.threads.max = 256

  • fs.s3a.connection.maximum = 1000

  • If your workload is small or your cluster resources are limited, consider lower values. Increase carefully if you need a higher throughput.

• Evaluate Committers

  • Start with the default (v1) committer and measure commit times in Spark UI.

  • If commit times become a bottleneck, consider the v2 directory committer, but plan for how you handle partial commits on failure.

• Tune Multipart Parameters

  • fs.s3a.multipart.size = 32 MB

  • fs.s3a.multipart.threshold = 320 MB

  • fs.s3a.fast.upload.active.blocks = 4

  • Validate your largest file sizes and watch for memory or concurrency issues.

• Monitor & Iterate

  • Observe performance metrics in Spark UI, VAST UI, and system metrics.

  • Adjust one parameter at a time to isolate impacts.

Understanding how executors are allocated in Apache Spark

Optimizing the allocation of executors can significantly improve workload performance.

• There are two primary approaches to setting up executors for a Spark job:

1. Static Allocation

2. Dynamic Allocation

Static Allocation

• Static allocation involves setting specific numbers of executors, cores, and memory in advance. To calculate these values effectively, compare the total resources on each worker (CPU cores and memory) to the resources requested per executor. For example, if each worker has 16 cores and 64 GB RAM, and you allocate 5 cores and 20 GB RAM per executor, you can run three executors per worker without over-allocating. This approach ensures each executor has sufficient resources without straining the system. These parameters remain fixed for the lifetime of the application.

Key Parameters:

•   --conf spark.executor.cores=5 \
    --conf spark.executor.memory=16g \
    --conf spark.executor.instances=3 \

• NOTE:
  Setting spark.executor.instances is optional because you can indirectly control the number of executors through spark.executor.cores and spark.executor.memory. By configuring these, you define the size of each executor, and how many can fit into your cluster depends on the total resources available on each worker node.

Advantages:

• Resource allocation is predictable, making it easier to manage in environments with fixed capacities.

• Simple to configure, as it doesn't require additional overhead to manage scaling.

Disadvantages:

• May lead to underutilization or overutilization of resources depending on the workload, as the number of executors does not change in response to the job's demands.

• Not ideal for varying workloads where the processing needs can change dramatically over time.

Dynamic Allocation

Dynamic allocation enables Spark to add or remove executors dynamically based on the workload. This means Spark can request more executors when processing demand is high and release them when demand decreases.

Key Parameters:

• spark.dynamicAllocation.enabled: Enables dynamic allocation.

• spark.dynamicAllocation.minExecutors: Minimum number of executors Spark will maintain.

• spark.dynamicAllocation.maxExecutors: Maximum number of executors Spark can allocate.

• spark.dynamicAllocation.initialExecutors: Initial number of executors Spark should start with.

• spark.dynamicAllocation.executorIdleTimeout: Duration after which an idle executor is removed.

Advantages:

• Improved Resource Utilization: Adjusts the number of executors based on the workload, potentially leading to better utilization of cluster resources.

Disadvantages:

• Complexity: Requires a more sophisticated setup, including proper configuration of the Spark cluster manager and possibly fine-tuning the parameters for optimal performance.

• Potential for Latency: Scaling decisions aren't instantaneous, which can introduce delays in executor provisioning, impacting job start times or scaling reactions to workload changes.

Choosing Between the Two

• Use Static Allocation when you have a predictable workload, or when operating in a static cluster environment.

• Use Dynamic Allocation for jobs with varying workloads, where resource utilization efficiency is critical. It's also beneficial when cluster workloads are unpredictable and diverse, making it hard to manually determine the optimal number of executors.

Therefore, we suggest using static allocation for benchmarking, as it is more predictable and "fair" for comparison.

Conclusion

Tuning S3A for optimal performance with VAST S3 involves balancing concurrency, network utilization, and memory constraints. By following the recommendations in this document—starting with conservative defaults, enabling VAST hard link, carefully increasing concurrency, and choosing appropriate multipart sizes—you can achieve high throughput and reliable performance.

Always remember to

1. Monitor your changes in a controlled manner.

2. Make sure your capacity planning (CPU, memory, network) aligns with the new concurrency levels.

3. Take a holistic view of performance across all workloads in the cluster.

If you have further questions or encounter unexpected behavior, please reach out to VAST Support with details about your workload, cluster configuration, and any relevant logs or metrics.