[MINOR][DOCS] Fix typos in comments and docstrings across multiple files

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala

@@ -81,7 +81,7 @@ private[v2] trait V2JDBCTest
     // nullable is true in the expectedSchema because Spark always sets nullable to true
     // regardless of the JDBC metadata https://github.com/apache/spark/pull/18445
     var expectedSchema = new StructType().add("ID", StringType, true, defaultMetadata())
-    // If function is not overriden we don't want to compare external engine types
+    // If function is not overridden we don't want to compare external engine types
     var expectedSchemaWithoutJdbcClientType =
       removeMetadataFromAllFields(expectedSchema, "jdbcClientType")
     var schemaWithoutJdbcClientType =
@@ -91,7 +91,7 @@ private[v2] trait V2JDBCTest
     t = spark.table(s"$catalogName.alt_table")
     expectedSchema = new StructType().add("ID", StringType, true, defaultMetadata())
 
-    // If function is not overriden we don't want to compare external engine types
+    // If function is not overridden we don't want to compare external engine types
     expectedSchemaWithoutJdbcClientType =
       removeMetadataFromAllFields(expectedSchema, "jdbcClientType")
     schemaWithoutJdbcClientType =
@@ -118,7 +118,7 @@ private[v2] trait V2JDBCTest
       .add("ID1", StringType, true, defaultMetadata())
       .add("ID2", StringType, true, defaultMetadata())
 
-    // If function is not overriden we don't want to compare external engine types
+    // If function is not overridden we don't want to compare external engine types
     val expectedSchemaWithoutJdbcClientType =
       removeMetadataFromAllFields(expectedSchema, "jdbcClientType")
     val schemaWithoutJdbcClientType =

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -2198,8 +2198,8 @@ private[spark] class DAGScheduler(
       log"shuffle checksum mismatch stage ${MDC(STAGE, mapStage)} in active jobs")
     val stagesCanRollback = filterAndAbortUnrollbackableStages(stagesToRollback)
 
-    // stages which cannot be rolled back were aborted which leads to removing the
-    // the dependant job(s) from the active jobs set, there could be no active jobs
+    // stages which cannot be rolled back were aborted which leads to removing
+    // the dependent job(s) from the active jobs set, there could be no active jobs
     // left depending on the indeterminate stage and hence no need to roll back any stages.
     val numActiveJobsWithStageAfterRollback =
       activeJobs.count(job => stagesToRollback.contains(job.finalStage))

python/benchmarks/bench_eval_type.py

@@ -140,7 +140,7 @@ def write_init_message(
     ) -> None:
         """Write the initial message with header, length + its data."""
 
-        # Write everything to a seperate buffer so we can
+        # Write everything to a separate buffer so we can
         # determine the length of the initial message.
         buf = io.BytesIO()
         cls.write_preamble(buf)

python/pyspark/pandas/generic.py

@@ -1882,7 +1882,7 @@ def std(
             where N represents the number of elements.
 
             .. versionchanged:: 3.4.0
-               Supported including arbitary integers.
+               Supported including arbitrary integers.
         numeric_only: bool, default None
             Include only float, int, boolean columns. False is not supported. This parameter
             is mainly for pandas compatibility.
@@ -1981,7 +1981,7 @@ def var(
             where N represents the number of elements.
 
             .. versionchanged:: 3.4.0
-               Supported including arbitary integers.
+               Supported including arbitrary integers.
         numeric_only: bool, default None
             Include only float, int, boolean columns. False is not supported. This parameter
             is mainly for pandas compatibility.
@@ -2213,7 +2213,7 @@ def sem(
             where N represents the number of elements.
 
             .. versionchanged:: 3.4.0
-               Supported including arbitary integers.
+               Supported including arbitrary integers.
         numeric_only: bool, default None
             Include only float, int, boolean columns. False is not supported. This parameter
             is mainly for pandas compatibility.

python/pyspark/pandas/groupby.py

@@ -851,7 +851,7 @@ def std(self, ddof: int = 1) -> FrameLike:
             where N represents the number of elements.
 
             .. versionchanged:: 3.4.0
-               Supported including arbitary integers.
+               Supported including arbitrary integers.
 
         Examples
         --------
@@ -986,7 +986,7 @@ def var(self, ddof: int = 1, numeric_only: bool = False) -> FrameLike:
             where N represents the number of elements.
 
             .. versionchanged:: 3.4.0
-               Supported including arbitary integers.
+               Supported including arbitrary integers.
 
         numeric_only : bool, default False
              Include only float, int, boolean columns.

python/pyspark/sql/tests/test_python_streaming_datasource.py

@@ -579,7 +579,7 @@ def test_stream_writer(self):
 
             # Test stream writer write and commit.
             # The first microbatch contain 30 rows and 2 partitions.
-            # Number of rows and partitions is writen by StreamWriter.commit().
+            # Number of rows and partitions is written by StreamWriter.commit().
             assertDataFrameEqual(self.spark.read.json(output_dir.name), [Row(2, 30)])
 
             self.spark.range(50, 80).repartition(2).write.format("json").mode("append").save(

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -856,7 +856,7 @@ private[spark] class Client(
    * The archive also contains some Spark configuration. Namely, it saves the contents of
    * SparkConf in a file to be loaded by the AM process.
    *
-   * @param confsToOverride configs that should overriden when creating the final spark conf file
+   * @param confsToOverride configs that should be overridden when creating the final spark conf file
    */
   private def createConfArchive(confsToOverride: Map[String, String]): File = {
     val hadoopConfFiles = new HashMap[String, File]()

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -343,7 +343,7 @@ private[yarn] class YarnAllocator(
       // config off if you only want YARN containers with the extra resources but Spark not to
       // schedule using them. Now for custom ResourceProfiles, it doesn't currently have a way
       // to only specify YARN resources without Spark scheduling off of them. This means for
-      // custom ResourceProfiles we propogate all the resources defined in the ResourceProfile
+      // custom ResourceProfiles we propagate all the resources defined in the ResourceProfile
       // to YARN. We still convert GPU and FPGA to the YARN build in types as well. This requires
       // that the name of any custom resources you specify match what they are defined as in YARN.
       val customResources = if (rp.id == DEFAULT_RESOURCE_PROFILE_ID) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ExpressionResolver.scala

@@ -1176,7 +1176,7 @@ class ExpressionResolver(
    *
    * -- GetViewColumnByNameAndOrdinal helps us to disambiguate the column names from different
    * -- tables by matching the same attribute names from those tables by their ordinal in the
-   * -- Project list, which is dependant on the order of tables in the inner join operator from
+   * -- Project list, which is dependent on the order of tables in the inner join operator from
    * -- the view plan:
    * -- getviewcolumnbynameandordinal(`spark_catalog`.`default`.`all_columns`, col1, 0, 2)
    * -- getviewcolumnbynameandordinal(`spark_catalog`.`default`.`all_columns`, col2, 0, 2)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HavingResolver.scala

@@ -91,7 +91,7 @@ class HavingResolver(resolver: Resolver, expressionResolver: ExpressionResolver)
    * Window [col1#0, sum(col1#0) windowspec...#2]
    * +- Aggregate [col1#0], [col1#0]
    *
-   * Since [[Filter]] is not dependant on the [[Window]], we push a [[Filter]] operator down to
+   * Since [[Filter]] is not dependent on the [[Window]], we push a [[Filter]] operator down to
    * the nearest [[Aggregate]] and perform the rest of resolution using its output:
    *
    * Window [col1#0, sum(col1#0) windowspec...#1]

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -2587,7 +2587,7 @@ case class Get(left: Expression, right: Expression)
 
   override def inputTypes: Seq[AbstractDataType] = left.dataType match {
     case _: ArrayType => Seq(ArrayType, IntegerType)
-    // Do not apply implicit cast if the first arguement is not array type.
+    // Do not apply implicit cast if the first argument is not array type.
     case _ => Nil
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/streaming/WriteToStreamStatement.scala

@@ -26,7 +26,7 @@ import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog}
 import org.apache.spark.sql.streaming.{OutputMode, Trigger}
 
 /**
- * A statement for Stream writing. It contains all neccessary param and will be resolved in the
+ * A statement for Stream writing. It contains all necessary param and will be resolved in the
  * rule [[ResolveStreamWrite]].
  *
  * @param userSpecifiedName  Query name optionally specified by the user.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CodeGenerationSuite.scala

@@ -206,7 +206,7 @@ class CodeGenerationSuite extends SparkFunSuite with ExpressionEvalHelper {
     }
   }
 
-  test("SPARK-22226: group splitted expressions into one method per nested class") {
+  test("SPARK-22226: group split expressions into one method per nested class") {
     val length = 10000
     val expressions = Seq.fill(length) {
       ToUTCTimestamp(

sql/core/src/main/scala/org/apache/spark/sql/execution/command/DataWritingCommand.scala

@@ -85,7 +85,7 @@ object DataWritingCommand {
    * or [[InsertIntoHiveTable]] command to write data, they both inherit metrics from
    * [[DataWritingCommand]], but after running [[InsertIntoHadoopFsRelationCommand]]
    * or [[InsertIntoHiveTable]], we only update metrics in these two command through
-   * [[BasicWriteJobStatsTracker]], we also need to propogate metrics to the command
+   * [[BasicWriteJobStatsTracker]], we also need to propagate metrics to the command
    * that actually calls [[InsertIntoHadoopFsRelationCommand]] or [[InsertIntoHiveTable]].
    *
    * @param sparkContext Current SparkContext.

sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala

@@ -80,7 +80,7 @@ class HadoopTableReader(
   //
   // In order keep consistency with Hive, we will let it be 0 in local mode also.
   private val _minSplitsPerRDD = if (sparkSession.sparkContext.isLocal) {
-    0 // will splitted based on block by default.
+    0 // will be split based on block by default.
   } else {
     math.max(hadoopConf.getInt("mapreduce.job.maps", 1),
       sparkSession.sparkContext.defaultMinPartitions)