diff --git a/python/pyspark/sql/session.py b/python/pyspark/sql/session.py
index e96eef79b87ec..b80d2dbaaaf00 100644
--- a/python/pyspark/sql/session.py
+++ b/python/pyspark/sql/session.py
@@ -1376,7 +1376,8 @@ def sql(self, sqlQuery: str, **kwargs: Any) -> DataFrame:
         if len(kwargs) > 0:
             sqlQuery = formatter.format(sqlQuery, **kwargs)
         try:
-            return DataFrame(self._jsparkSession.sql(sqlQuery), self)
+            args = formatter.unused_args
+            return DataFrame(self._jsparkSession.sql(sqlQuery, args), self)
         finally:
             if len(kwargs) > 0:
                 formatter.clear()
diff --git a/python/pyspark/sql/sql_formatter.py b/python/pyspark/sql/sql_formatter.py
index 5e79b9ff5ea98..9c669fa8091d8 100644
--- a/python/pyspark/sql/sql_formatter.py
+++ b/python/pyspark/sql/sql_formatter.py
@@ -37,6 +37,7 @@ class SQLStringFormatter(string.Formatter):
     def __init__(self, session: "SparkSession") -> None:
         self._session: "SparkSession" = session
         self._temp_views: List[Tuple[DataFrame, str]] = []
+        self.unused_args: Mapping[str, str] = {}
 
     def get_field(self, field_name: str, args: Sequence[Any], kwargs: Mapping[str, Any]) -> Any:
         obj, first = super(SQLStringFormatter, self).get_field(field_name, args, kwargs)
@@ -78,7 +79,16 @@ def _convert_value(self, val: Any, field_name: str) -> Optional[str]:
         else:
             return val
 
+    def check_unused_args(
+        self, used_args: Sequence[Any], args: Sequence[Any], kwargs: Mapping[str, Any]
+    ) -> None:
+        for k in kwargs:
+            if isinstance(kwargs[k], str) and not used_args.__contains__(k):
+                self.unused_args[k] = kwargs[k]
+        return super().check_unused_args(used_args, args, kwargs)
+
     def clear(self) -> None:
         for _, n in self._temp_views:
             self._session.catalog.dropTempView(n)
         self._temp_views = []
+        self.unused_args = []