code: def test_key_by_extracts_keys_and_keeps_values_untouched(self):
      col = self.sc.parallelize(
          ["key1_value1", "key1_value2", "key2_value1"]
      )
  
      def underscore_separated_key_extractor(el):
          return el.split("_")[0]
  
      result = composite_funcs.key_by(
          self.backend,
          col,
          underscore_separated_key_extractor,
          stage_name="Key by",
      ).collect()
  
      self.assertSetEqual(
          {
              ("key1", "key1_value1"),
              ("key1", "key1_value2"),
              ("key2", "key2_value1"),
          },
          set(result),
      )
  
imports: import sys
  import unittest
  from dataclasses import dataclass
  from typing import List
  import apache_beam
  import apache_beam.testing.test_pipeline
  import apache_beam.testing.util
  from absl.testing import parameterized
  import pipeline_dp
  from pipeline_dp import pipeline_functions
  from pipeline_dp import pipeline_backend
label_desc: Use image node_id 2 for calling the SparkRDDBackendTest obj's underlying member method code with example usage: obj.test_key_by_extracts_keys_and_keeps_values_untouched() and returns: el