code: def test_key_by_extracts_keys_and_keeps_values_untouched(self):
      with test_pipeline.TestPipeline() as p:
          col = p | beam.Create(
              ["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",
          )
  
          beam_util.assert_that(
              result,
              beam_util.equal_to(
                  {
                      ("key1", "key1_value1"),
                      ("key1", "key1_value2"),
                      ("key2", "key2_value1"),
                  }
              ),
          )
  
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 BeamBackendTest obj's underlying member method code with example usage: obj.test_key_by_extracts_keys_and_keeps_values_untouched() and returns: el