code: def key_by(self, key_selector):
          
          op = Operator(
              _generate_uuid(),
              OpType.KeyBy,
              "KeyBy",
              other=key_selector,
              num_instances=self.env.config.parallelism)
          return self.__register(op)
docs: Applies a key_by operator to the stream.
  
          Attributes:
               key_attribute_index (int): The index of the key attributed
               (assuming tuple records).