code: def testQuantileBuckets(self, input_dtype):
      def analyzer_fn(inputs):
          return {
              "q_b": tft.quantiles(
                  tf.cast(inputs["x"], input_dtype),
                  num_buckets=3,
                  epsilon=0.00001,
              )
          }
  
      # NOTE: We force 3 batches: data has 3000 elements and we request a batch
      # size of 1000.
      input_data = [{"x": [x]} for x in range(1, 3000)]
      input_metadata = tft.DatasetMetadata.from_feature_spec(
          {
              "x": tf.io.FixedLenFeature(
                  [1], tft_unit.canonical_numeric_dtype(input_dtype)
              )
          }
      )
      # The expected data has 2 boundaries that divides the data into 3 buckets.
      expected_outputs = {"q_b": np.array([[1000, 2000]], np.float32)}
      self.assertAnalyzerOutputs(
          input_data,
          input_metadata,
          analyzer_fn,
          expected_outputs,
          desired_batch_size=1000,
      )
  
imports: import contextlib
  import random
  import numpy
  import tensorflow
  import tensorflow_transform
  from tensorflow_transform import analyzers
  from tensorflow_transform.beam import impl
  from tensorflow_transform.beam import tft_unit
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 8 for calling the BucketizeIntegrationTest obj's underlying member method code with example usage: obj.testQuantileBuckets(input_dtype) and returns: dict