code: def test_bucketed_write(self):
      data = [
          (1, "foo", 3.0),
          (2, "foo", 5.0),
          (3, "bar", -1.0),
          (4, "bar", 6.0),
      ]
      df = self.spark.createDataFrame(data, ["x", "y", "z"])
  
      def count_bucketed_cols(names, table="pyspark_bucket"):
          """Given a sequence of column names and a table name
          query the catalog and return number o columns which are
          used for bucketing
          """
          cols = self.spark.catalog.listColumns(table)
          num = len([c for c in cols if c.name in names and c.isBucket])
          return num
  
      with self.table("pyspark_bucket"):
          # Test write with one bucketing column
          df.write.bucketBy(3, "x").mode("overwrite").saveAsTable(
              "pyspark_bucket"
          )
          self.assertEqual(count_bucketed_cols(["x"]), 1)
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
          # Test write two bucketing columns
          df.write.bucketBy(3, "x", "y").mode("overwrite").saveAsTable(
              "pyspark_bucket"
          )
          self.assertEqual(count_bucketed_cols(["x", "y"]), 2)
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
          # Test write with bucket and sort
          df.write.bucketBy(2, "x").sortBy("z").mode(
              "overwrite"
          ).saveAsTable("pyspark_bucket")
          self.assertEqual(count_bucketed_cols(["x"]), 1)
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
          # Test write with a list of columns
          df.write.bucketBy(3, ["x", "y"]).mode(
              "overwrite"
          ).saveAsTable("pyspark_bucket")
          self.assertEqual(count_bucketed_cols(["x", "y"]), 2)
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
          # Test write with bucket and sort with a list of columns
          (
              df.write.bucketBy(2, "x")
              .sortBy(["y", "z"])
              .mode("overwrite")
              .saveAsTable("pyspark_bucket")
          )
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
          # Test write with bucket and sort with multiple columns
          (
              df.write.bucketBy(2, "x")
              .sortBy("y", "z")
              .mode("overwrite")
              .saveAsTable("pyspark_bucket")
          )
          self.assertSetEqual(
              set(data),
              set(self.spark.table("pyspark_bucket").collect()),
          )
  
imports: import os
  import shutil
  import tempfile
  from pyspark.errors import AnalysisException
  from pyspark.sql.functions import col
  from pyspark.sql.readwriter import DataFrameWriterV2
  from pyspark.sql.types import StructType, StructField, StringType
  from pyspark.testing.sqlutils import ReusedSQLTestCase
label_desc: Use image node_id 3 for calling the ReadwriterTestsMixin obj's underlying member method code with example usage: obj.test_bucketed_write() and returns: num