如何在Apache Beam Dataflow作业中检测CSV重复值并写入BigQuery

Hey there! Since you're new to Apache Beam, let's walk through adding duplicate detection to your existing pipeline step by step. First, we need to clarify what counts as a "duplicate" for your use case—usually this is based on business-specific unique keys (like a user ID, order number, or combination of fields) rather than the entire record (since your filename and load_dt fields will be consistent per run).

Option 1: Remove Duplicates (Keep Only Unique Records) #

If your goal is to load only unique records into BigQuery, use beam.Distinct with a custom key function to define what makes a record unique.

Step 1: Define Your Unique Key Function #

First, add a function that extracts the unique identifier(s) from your transformed record. For example, if duplicates are based on email and order_id:

def get_unique_key(record):
    # Replace with your actual unique key fields (single field or tuple of fields)
    return (record['email'], record['order_id'])

Step 2: Add the Deduplication Step to Your Pipeline #

Insert this step right after converting rows to BigQuery dictionaries, before writing to BigQuery:

with beam.Pipeline(options=p_opts) as pipeline:
    rows = pipeline | "Read from text file" >> beam.io.ReadFromText(known_args.input)

    dict_records = rows | "Convert to BigQuery row" >> beam.Map(
        lambda r: row_transformer.parse(r))

    # Add deduplication here
    unique_records = dict_records | "Remove duplicates by unique key" >> beam.Distinct(get_unique_key)

    # Write only unique records to BigQuery
    unique_records | "Write to BigQuery" >> beam.io.Write(
        beam.io.BigQuerySink(known_args.output,
                             create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
                             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

Option 2: Detect and Log Duplicates (Keep Uniques + Track Duplicates) #

If you need to identify duplicates and store them separately (e.g., in a "duplicates" BigQuery table for later review), use GroupByKey to count occurrences of each unique key.

Step 1: Update Your Pipeline to Split Uniques and Duplicates #

Modify your pipeline to split records into two branches: unique records (written to your main table) and duplicate records (written to a separate table):

with beam.Pipeline(options=p_opts) as pipeline:
    rows = pipeline | "Read from text file" >> beam.io.ReadFromText(known_args.input)

    dict_records = rows | "Convert to BigQuery row" >> beam.Map(
        lambda r: row_transformer.parse(r))

    # 1. Tag records with their unique key
    keyed_records = dict_records | "Tag with unique key" >> beam.WithKeys(get_unique_key)

    # 2. Group records by their unique key
    grouped_records = keyed_records | "Group by unique key" >> beam.GroupByKey()

    # 3. Split into unique and duplicate records
    def split_unique_duplicates(key_value):
        key, records = key_value
        record_list = list(records)
        if len(record_list) > 1:
            # Mark duplicates and send to a separate output branch
            for record in record_list:
                record['is_duplicate'] = True
                yield beam.pvalue.TaggedOutput('duplicates', record)
        else:
            # Mark as unique and send to main output
            record_list[0]['is_duplicate'] = False
            yield record_list[0]

    # Use tagged outputs to separate the two record types
    results = grouped_records | "Split unique/duplicate records" >> beam.FlatMap(
        split_unique_duplicates).with_outputs('duplicates', main='unique')

    # 4. Write unique records to your main BigQuery table
    results.unique | "Write unique records to BigQuery" >> beam.io.Write(
        beam.io.BigQuerySink(known_args.output,
                             create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
                             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

    # 5. Write duplicates to a dedicated "duplicates" table
    results.duplicates | "Write duplicates to BigQuery" >> beam.io.Write(
        beam.io.BigQuerySink(
            f"{known_args.output}_duplicates",  # Append "_duplicates" to your main table name
            # Update schema to include the new is_duplicate field
            schema=f"{known_args.fields},is_duplicate:BOOLEAN,filename:STRING,load_dt:DATE",
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

Key Notes to Remember #

• Define Your Unique Key Clearly: Make sure get_unique_key uses fields that truly identify a unique record for your business. If you need to check for full record duplicates (rare, given your filename/load_dt fields), skip the key function and use beam.Distinct() directly.
• Schema Updates: If adding the is_duplicate field, ensure your BigQuery schema (for both main and duplicates tables) includes this boolean field.
• Test with Small Data: Run your pipeline on a small sample first to verify the deduplication logic works as expected.

内容的提问来源于stack exchange，提问作者J.C Guzman