[Experimental] Add FilterComponent for row-level example filtering

[vkarampudi]

Description

This PR introduces an experimental FilterComponent to TFX, addressing a long-standing feature request (Issue #6909) for row-level example filtering within pipelines.

Currently, TFX has no built-in component to filter anomalous or unwanted examples after ingestion. The Transform component (TFT) is strictly map-only and does not support dropping rows. As a result, users have had to write custom Apache Beam code or perform out-of-band pre-processing. The new FilterComponent provides a clean, native, and reusable way to filter datasets split-by-split using a user-defined Python function.

Proposed Changes

1. Component Definition (component.py):
   • Created FilterSpec and FilterComponent.
   • Accepts an input Examples channel and a string import path to a Python filter function (filter_fn_path).
   • Outputs a filtered_examples Examples channel.
2. Beam Executor (executor.py):
   • Implements a high-performance Apache Beam pipeline.
   • Automatically decodes all splits present in the input Examples artifact (e.g., train, eval).
   • Dynamically imports the user-defined filter function and applies beam.Filter.
   • Writes the output splits using standard TFRecord Gzip compression while preserving all metadata (spans, versions, split names) for downstream components.
3. Tests:
   • component_test.py: Verifies component instantiation and spec parsing.
   • executor_test.py: Runs a full integration test with simulated train and eval splits, executing the Beam filter pipeline and verifying correct output records.

Usage Example

# 1. Define a filter function in a shared module (e.g., my_filters.py)
def age_filter_fn(serialized_example: bytes) -> bool:
  example = tf.train.Example()
  example.ParseFromString(serialized_example)
  age_feature = example.features.feature.get('age')
  if age_feature:
    return age_feature.int64_list.value[0] > 18
  return False

# 2. Add the FilterComponent to your pipeline
from tfx.components.experimental.filter.component import FilterComponent

filter_data = FilterComponent(
    examples=example_gen.outputs['examples'],
    filter_fn_path='my_filters.age_filter_fn'
)

# 3. Pass the filtered output to Transform/Trainer
transform = Transform(
    examples=filter_data.outputs['filtered_examples'],
    schema=schema_gen.outputs['schema'],
    preprocessing_fn=...
)

[rwitcher]

LGTM

[pritamdodeja]

Wow! Thank you for implementing this! Reviewed the code/approach and it is exactly what I imagined for the tf.train.Example case. Do you think this component would also work with tf.train.SequenceExample also? What I mean is that the filter criteria is still a non-sequence type, but the record itself can have sequences. For example, imagine a data quality threshold of some sort. Another scenario which I remember implementing in a likely hacky way was polygon based filtering, where you want to crop data using regions in space. Classic example would be filter data from the "ocean" in the taxicab dataset. In any case, super happy to see this get merged!