AWS clean-rooms documentation change
Summary
Updated Spark configuration documentation with new properties, default value changes (uppercase to lowercase), added configuration limits (max 50 properties, 500 chars each), and corrected default values for several Spark properties including retry settings, compression, and timezone.
Security assessment
The changes are routine documentation updates for Spark configuration properties in AWS Clean Rooms ML input channels. No security vulnerabilities, incidents, or security features are mentioned. Changes include formatting corrections (TRUE/FALSE to true/false), added property limits for operational management, and new Spark configuration options for performance tuning. The change to UTC timezone is for consistency, not security. The retry configuration (spark.hadoop.fs.s3.maxRetries) is a standard reliability feature, not a security fix.
Diff
diff --git a/clean-rooms/latest/userguide/create-ml-input-channel.md b/clean-rooms/latest/userguide/create-ml-input-channel.md index ebfc30634..d7d5afa1f 100644 --- a//clean-rooms/latest/userguide/create-ml-input-channel.md +++ b//clean-rooms/latest/userguide/create-ml-input-channel.md @@ -87 +87 @@ We recommend avoiding values in the input dataset that are associated with only -The following table provides a definition for each property. +The following tables provide a definition for each property. @@ -90,0 +91,4 @@ For more information about Spark properties, see [Spark Properties](https://spar +###### Note + +You can configure a maximum of 50 Spark properties. Each property value can be up to 500 characters. + @@ -95 +99 @@ spark.sql.files.maxPartitionBytes | Sets the maximum number of bytes to pack in -spark.hadoop.fs.s3.maxRetries | Sets the maximum number of retry attempts for Amazon S3 file operations. | +spark.hadoop.fs.s3.maxRetries | Sets the maximum number of retry attempts for Amazon S3 file operations. | (none) @@ -102,2 +106,10 @@ spark.network.timeout | Sets the default timeout for all network interactions. -spark.rdd.compress | Specifies whether to compress serialized RDD partitions using spark.io.compression.codec. Applies to StorageLevel.MEMORY_ONLY_SER in Java and Scala, or StorageLevel.MEMORY_ONLY in Python. Reduces storage space but requires additional CPU processing time. | FALSE -spark.shuffle.spill.compress | Specifies whether to compress shuffle spill data using spark.io.compression.codec. | TRUE +spark.rdd.compress | Specifies whether to compress serialized RDD partitions using spark.io.compression.codec. Applies to StorageLevel.MEMORY_ONLY_SER in Java and Scala, or StorageLevel.MEMORY_ONLY in Python. Reduces storage space but requires additional CPU processing time. | false +spark.shuffle.spill.compress | Specifies whether to compress shuffle spill data using spark.io.compression.codec. | true +spark.shuffle.compress | Specifies whether to compress map output files. Compression uses spark.io.compression.codec. | true +spark.shuffle.service.index.cache.size | Sets the cache size limit, in bytes unless otherwise specified. | 100m +spark.shuffle.io.maxRetries | Sets the maximum number of retries for fetches that fail due to IO-related exceptions. | 3 +spark.shuffle.io.retryWait | Sets the wait time between retries of fetches. The maximum delay caused by retrying is 15 seconds by default, calculated as maxRetries * retryWait. | 5s +spark.shuffle.io.connectionTimeout | Sets the timeout for established connections between shuffle servers and clients to be marked as idle and closed if there are still outstanding fetch requests but no traffic on the channel. | (value of spark.network.timeout) +spark.driver.maxResultSize | Sets the total size limit of serialized results of all partitions for each Spark action, in bytes. Should be at least 1M, or 0 for unlimited. | 1g +spark.memory.fraction | Sets the fraction of (heap space - 300MB) used for execution and storage. The lower this value, the more frequently spills and cached data eviction occur. Leaving this at the default value is recommended. | 0.6 +spark.scheduler.mode | Sets the scheduling mode between jobs submitted to the same SparkContext. Can be set to FAIR to use fair sharing instead of queueing jobs one after another. Supported values: FAIR, FIFO. | FIFO @@ -106 +118 @@ spark.sql.adaptive.autoBroadcastJoinThreshold | Sets the maximum table size in -spark.sql.adaptive.coalescePartitions.enabled | Specifies whether to coalesce contiguous shuffle partitions based on spark.sql.adaptive.advisoryPartitionSizeInBytes to optimize task size. Requires spark.sql.adaptive.enabled to be true. | TRUE +spark.sql.adaptive.coalescePartitions.enabled | Specifies whether to coalesce contiguous shuffle partitions based on spark.sql.adaptive.advisoryPartitionSizeInBytes to optimize task size. Requires spark.sql.adaptive.enabled to be true. | true @@ -109,4 +121,4 @@ spark.sql.adaptive.coalescePartitions.minPartitionSize | Sets the minimum size -spark.sql.adaptive.coalescePartitions.parallelismFirst | Specifies whether to calculate partition sizes based on cluster parallelism instead of spark.sql.adaptive.advisoryPartitionSizeInBytes during partition coalescing. Generates smaller partition sizes than the configured target size to maximize parallelism. We recommend setting this to false on busy clusters to improve resource utilization by preventing excessive small tasks. | TRUE -spark.sql.adaptive.enabled | Specifies whether to enable adaptive query execution to re-optimize query plans during query execution, based on accurate runtime statistics. | TRUE -spark.sql.adaptive.forceOptimizeSkewedJoin | Specifies whether to force enable OptimizeSkewedJoin even if it introduces extra shuffle. | FALSE -spark.sql.adaptive.localShuffleReader.enabled | Specifies whether to use local shuffle readers when shuffle partitioning isn't required, such as after converting from sort-merge joins to broadcast-hash joins. Requires spark.sql.adaptive.enabled to be true. | TRUE +spark.sql.adaptive.coalescePartitions.parallelismFirst | Specifies whether to calculate partition sizes based on cluster parallelism instead of spark.sql.adaptive.advisoryPartitionSizeInBytes during partition coalescing. Generates smaller partition sizes than the configured target size to maximize parallelism. We recommend setting this to false on busy clusters to improve resource utilization by preventing excessive small tasks. | true +spark.sql.adaptive.enabled | Specifies whether to enable adaptive query execution to re-optimize query plans during query execution, based on accurate runtime statistics. | true +spark.sql.adaptive.forceOptimizeSkewedJoin | Specifies whether to force enable OptimizeSkewedJoin even if it introduces extra shuffle. | false +spark.sql.adaptive.localShuffleReader.enabled | Specifies whether to use local shuffle readers when shuffle partitioning isn't required, such as after converting from sort-merge joins to broadcast-hash joins. Requires spark.sql.adaptive.enabled to be true. | true @@ -117 +129 @@ Overrides spark.sql.join.preferSortMergeJoin setting. | 0 bytes -spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled | Specifies whether to optimize skewed shuffle partitions by splitting them into smaller partitions based on spark.sql.adaptive.advisoryPartitionSizeInBytes. Requires spark.sql.adaptive.enabled to be true. | TRUE +spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled | Specifies whether to optimize skewed shuffle partitions by splitting them into smaller partitions based on spark.sql.adaptive.advisoryPartitionSizeInBytes. Requires spark.sql.adaptive.enabled to be true. | true @@ -119 +131 @@ spark.sql.adaptive.rebalancePartitionsSmallPartitionFactor | Defines the size t -spark.sql.adaptive.skewJoin.enabled | Specifies whether to handle data skew in shuffled joins by splitting and optionally replicating skewed partitions. Applies to sort-merge and shuffled hash joins. Requires spark.sql.adaptive.enabled to be true. | TRUE +spark.sql.adaptive.skewJoin.enabled | Specifies whether to handle data skew in shuffled joins by splitting and optionally replicating skewed partitions. Applies to sort-merge and shuffled hash joins. Requires spark.sql.adaptive.enabled to be true. | true @@ -128 +139,0 @@ We recommend setting this value larger than spark.sql.adaptive.advisoryPartition -spark.sql.autoBroadcastJoinThreshold | Sets the maximum table size in bytes for broadcasting to worker nodes during joins. Set to -1 to disable broadcasting. | 10MB @@ -130,2 +141,2 @@ spark.sql.broadcastTimeout | Controls the timeout period in seconds for the bro -spark.sql.cbo.enabled | Specifies whether to enable cost-based optimization (CBO) for plan statistics estimation. | FALSE -spark.sql.cbo.joinReorder.dp.star.filter | Specifies whether to apply star-join filter heuristics during cost-based join enumeration. | FALSE +spark.sql.cbo.enabled | Specifies whether to enable cost-based optimization (CBO) for plan statistics estimation. | false +spark.sql.cbo.joinReorder.dp.star.filter | Specifies whether to apply star-join filter heuristics during cost-based join enumeration. | false @@ -133,3 +144,3 @@ spark.sql.cbo.joinReorder.dp.threshold | Sets the maximum number of joined node -spark.sql.cbo.joinReorder.enabled | Specifies whether to enable join reordering in cost-based optimization (CBO). | FALSE -spark.sql.cbo.planStats.enabled | Specifies whether to fetch row counts and column statistics from the catalog during logical plan generation. | FALSE -spark.sql.cbo.starSchemaDetection | Specifies whether to enable join reordering based on star schema detection. | FALSE +spark.sql.cbo.joinReorder.enabled | Specifies whether to enable join reordering in cost-based optimization (CBO). | false +spark.sql.cbo.planStats.enabled | Specifies whether to fetch row counts and column statistics from the catalog during logical plan generation. | false +spark.sql.cbo.starSchemaDetection | Specifies whether to enable join reordering based on star schema detection. | false @@ -140,12 +151,12 @@ spark.sql.inMemoryColumnarStorage.batchSize | Controls the batch size for colum -spark.sql.inMemoryColumnarStorage.compressed | Specifies whether to automatically select compression codecs for columns based on data statistics. | TRUE -spark.sql.inMemoryColumnarStorage.enableVectorizedReader | Specifies whether to enable vectorized reading for columnar caching. | TRUE -spark.sql.legacy.allowHashOnMapType | Specifies whether to allow hash operations on map type data structures. This legacy setting maintains compatibility with older Spark versions' map type handling. | -spark.sql.legacy.allowNegativeScaleOfDecimal | Specifies whether to allow negative scale values in decimal type definitions. This legacy setting maintains compatibility with older Spark versions that supported negative decimal scales. | -spark.sql.legacy.castComplexTypesToString.enabled | Specifies whether to enable legacy behavior for casting complex types to strings. Maintains compatibility with older Spark versions' type conversion rules. | -spark.sql.legacy.charVarcharAsString | Specifies whether to treat CHAR and VARCHAR types as STRING types. This legacy setting provides compatibility with older Spark versions' string type handling. | -spark.sql.legacy.createEmptyCollectionUsingStringType | Specifies whether to create empty collections using string type elements. This legacy setting maintains compatibility with older Spark versions' collection initialization behavior. | -spark.sql.legacy.exponentLiteralAsDecimal.enabled | Specifies whether to interpret exponential literals as decimal types. This legacy setting maintains compatibility with older Spark versions' numeric literal handling. | -spark.sql.legacy.json.allowEmptyString.enabled | Specifies whether to allow empty strings in JSON processing. This legacy setting maintains compatibility with older Spark versions' JSON parsing behavior. | -spark.sql.legacy.parquet.int96RebaseModelRead | Specifies whether to use legacy INT96 timestamp rebase mode when reading Parquet files. This legacy setting maintains compatibility with older Spark versions' timestamp handling. | -spark.sql.legacy.timeParserPolicy | Controls the time parsing behavior for backwards compatibility. This legacy setting determines how timestamps and dates are parsed from strings. | -spark.sql.legacy.typeCoercion.datetimeToString.enabled | Specifies whether to enable legacy type coercion behavior when converting datetime values to strings. Maintains compatibility with older Spark versions' datetime conversion rules. | +spark.sql.inMemoryColumnarStorage.compressed | Specifies whether to automatically select compression codecs for columns based on data statistics. | true +spark.sql.inMemoryColumnarStorage.enableVectorizedReader | Specifies whether to enable vectorized reading for columnar caching. | true +spark.sql.legacy.allowHashOnMapType | Specifies whether to allow hash operations on map type data structures. This legacy setting maintains compatibility with older Spark versions' map type handling. | (none) +spark.sql.legacy.allowNegativeScaleOfDecimal | Specifies whether to allow negative scale values in decimal type definitions. This legacy setting maintains compatibility with older Spark versions that supported negative decimal scales. | (none) +spark.sql.legacy.castComplexTypesToString.enabled | Specifies whether to enable legacy behavior for casting complex types to strings. Maintains compatibility with older Spark versions' type conversion rules. | (none) +spark.sql.legacy.charVarcharAsString | Specifies whether to treat CHAR and VARCHAR types as STRING types. This legacy setting provides compatibility with older Spark versions' string type handling. | (none) +spark.sql.legacy.createEmptyCollectionUsingStringType | Specifies whether to create empty collections using string type elements. This legacy setting maintains compatibility with older Spark versions' collection initialization behavior. | (none) +spark.sql.legacy.exponentLiteralAsDecimal.enabled | Specifies whether to interpret exponential literals as decimal types. This legacy setting maintains compatibility with older Spark versions' numeric literal handling. | (none) +spark.sql.legacy.json.allowEmptyString.enabled | Specifies whether to allow empty strings in JSON processing. This legacy setting maintains compatibility with older Spark versions' JSON parsing behavior. | (none) +spark.sql.legacy.parquet.int96RebaseModelRead | Specifies whether to use legacy INT96 timestamp rebase mode when reading Parquet files. This legacy setting maintains compatibility with older Spark versions' timestamp handling. | (none) +spark.sql.legacy.timeParserPolicy | Controls the time parsing behavior for backwards compatibility. This legacy setting determines how timestamps and dates are parsed from strings. | (none) +spark.sql.legacy.typeCoercion.datetimeToString.enabled | Specifies whether to enable legacy type coercion behavior when converting datetime values to strings. Maintains compatibility with older Spark versions' datetime conversion rules. | (none) @@ -159,3 +170,3 @@ spark.sql.metadataCacheTTLSeconds | Controls the time-to-live (TTL) for metadat -spark.sql.optimizer.collapseProjectAlwaysInline | Specifies whether to collapse adjacent projections and inline expressions, even when it causes duplication. | FALSE -spark.sql.optimizer.dynamicPartitionPruning.enabled | Specifies whether to generate predicates for partition columns used as join keys. | TRUE -spark.sql.optimizer.enableCsvExpressionOptimization | Specifies whether to optimize CSV expressions in SQL optimizer by pruning unnecessary columns from from_csv operations. | TRUE +spark.sql.optimizer.collapseProjectAlwaysInline | Specifies whether to collapse adjacent projections and inline expressions, even when it causes duplication. | false +spark.sql.optimizer.dynamicPartitionPruning.enabled | Specifies whether to generate predicates for partition columns used as join keys. | true +spark.sql.optimizer.enableCsvExpressionOptimization | Specifies whether to optimize CSV expressions in SQL optimizer by pruning unnecessary columns from from_csv operations. | true @@ -166 +177 @@ spark.sql.optimizer.enableJsonExpressionOptimization | Specifies whether to opt -| TRUE +| true @@ -170 +181 @@ spark.sql.optimizer.runtime.bloomFilter.creationSideThreshold | Defines the max -spark.sql.optimizer.runtime.bloomFilter.enabled | Specifies whether to insert a Bloom filter to reduce shuffle data when one side of a shuffle join has a selective predicate. | TRUE +spark.sql.optimizer.runtime.bloomFilter.enabled | Specifies whether to insert a Bloom filter to reduce shuffle data when one side of a shuffle join has a selective predicate. | true @@ -174 +184,0 @@ spark.sql.optimizer.runtime.bloomFilter.maxNumItems | Sets the maximum number o -spark.sql.optimizer.runtime.bloomFilter.number.threshold | Limits the maximum number of non-DPP runtime filters allowed per query to prevent out-of-memory errors in the driver. | 10 @@ -183,2 +193,3 @@ Limitations: -| TRUE -spark.sql.optimizer.runtimeFilter.semiJoinReduction.enabled | Specifies whether to insert a semi-join to reduce shuffle data when one side of a shuffle join has a selective predicate. | FALSE +| true +spark.sql.optimizer.runtimeFilter.number.threshold | Sets the total number of injected runtime filters (non-DPP). This is to prevent driver OOMs with too many Bloom filters. | 10 +spark.sql.optimizer.runtimeFilter.semiJoinReduction.enabled | Specifies whether to insert a semi-join to reduce shuffle data when one side of a shuffle join has a selective predicate. | false @@ -188 +199 @@ spark.sql.parquet.aggregatePushdown | Specifies whether to push down aggregates -Throws an exception if statistics are missing from any Parquet file footer. | FALSE +Throws an exception if statistics are missing from any Parquet file footer. | false @@ -189,0 +201,17 @@ spark.sql.parquet.columnarReaderBatchSize | Controls the number of rows in each +spark.sql.parquet.enableVectorizedReader | Specifies whether to enable vectorized Parquet decoding. | true +spark.sql.shuffle.partitions | Sets the default number of partitions for data shuffling during joins or aggregations. Cannot be modified between structured streaming query restarts from the same checkpoint location. | 200 +spark.sql.shuffledHashJoinFactor | Defines the multiplication factor used to determine shuffle hash join eligibility. A shuffle hash join is selected when the small-side data size multiplied by this factor is less than the large-side data size. | 3 +spark.sql.sources.parallelPartitionDiscovery.threshold | Sets the maximum number of paths for driver-side file listing with file-based sources (Parquet, JSON, and ORC). When exceeded during partition discovery, files are listed using a separate Spark distributed job. | 32 +spark.sql.statistics.histogram.enabled | Specifies whether to generate equi-height histograms during column statistics computation to improve estimation accuracy. Requires an additional table scan beyond the one needed for basic column statistics. | false +spark.dynamicAllocation.executorIdleTimeout | Sets the duration an executor must be idle before it is removed when dynamic allocation is enabled. | 60s +spark.dynamicAllocation.schedulerBacklogTimeout | Sets the duration that pending tasks must be backlogged before new executors are requested when dynamic allocation is enabled. | 1s +spark.dynamicAllocation.sustainedSchedulerBacklogTimeout | Same as spark.dynamicAllocation.schedulerBacklogTimeout, but used only for subsequent executor requests. | (value of spark.dynamicAllocation.schedulerBacklogTimeout) +spark.scheduler.minRegisteredResourcesRatio | Sets the minimum ratio of registered resources (registered resources / total expected resources) to wait for before scheduling begins. Specified as a double between 0.0 and 1.0. Regardless of whether the minimum ratio of resources has been reached, the maximum amount of time it will wait before scheduling begins is controlled by spark.scheduler.maxRegisteredResourcesWaitingTime. | 0.8 +spark.scheduler.maxRegisteredResourcesWaitingTime | Sets the maximum amount of time to wait for resources to register before scheduling begins. | 30s +spark.sql.hive.metastorePartitionPruningFallbackOnException | Specifies whether to fall back to getting all partitions from Hive metastore and perform partition pruning on the Spark client side when encountering MetaException from the metastore. | false + +Property Name | Description | Default Value +---|---|--- +spark.sql.autoBroadcastJoinThreshold | Sets the maximum table size in bytes for broadcasting to worker nodes during joins. Set to -1 to disable broadcasting. | 10MB (-1 for CR.4X with 32 workers) +spark.dynamicAllocation.enabled | Specifies whether to use dynamic resource allocation, which scales the number of executors registered with this application up and down based on the workload. | true +spark.io.compression.codec | Sets the codec used to compress internal data such as RDD partitions, event log, broadcast variables, and shuffle outputs. Supported values: lz4, snappy, zstd, gzip. | snappy @@ -194,5 +222 @@ spark.sql.session.timeZone | Defines the session time zone for handling timesta -| (value of local timezone) -spark.sql.shuffle.partitions | Sets the default number of partitions for data shuffling during joins or aggregations. Cannot be modified between structured streaming query restarts from the same checkpoint location. | 200 -spark.sql.shuffledHashJoinFactor | Defines the multiplication factor used to determine shuffle hash join eligibility. A shuffle hash join is selected when the small-side data size multiplied by this factor is less than the large-side data size. | 3 -spark.sql.sources.parallelPartitionDiscovery.threshold | Sets the maximum number of paths for driver-side file listing with file-based sources (Parquet, JSON, and ORC). When exceeded during partition discovery, files are listed using a separate Spark distributed job. | 32 -spark.sql.statistics.histogram.enabled | Specifies whether to generate equi-height histograms during column statistics computation to improve estimation accuracy. Requires an additional table scan beyond the one needed for basic column statistics. | FALSE +| UTC