使用MongoDB聚合与Lookup在Golang中查找重复数据

Hey there! Let's figure out how to find duplicate Event documents using MongoDB's aggregation framework (including $lookup) in Go. Based on your Event struct definition, I'll walk you through a practical solution that works for common duplicate scenarios (like duplicates on old_id, name, or slug).

Core Approach #

The idea is to use MongoDB's aggregation pipeline to:

1. Group documents by the field you suspect has duplicates, counting occurrences and collecting document IDs.
2. Filter groups to only keep those with more than one document (actual duplicates).
3. Lookup full documents using the collected IDs to get the complete details of each duplicate entry.

Step-by-Step Implementation #

1. Setup Dependencies #

First, make sure you have the official MongoDB Go driver installed:

go get go.mongodb.org/mongo-driver/mongo

2. Aggregation Pipeline Function #

Here's a reusable function that finds duplicates based on a specified field:

import (
    "context"
    "fmt"
    "go.mongodb.org/mongo-driver/bson"
    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
)

// Event matches your provided struct definition
type Event struct {
    ID       string `bson:"_id" json:"_id" valid:"alphanum,printableascii"`
    OldID    string `bson:"old_id" json:"old_id" valid:"alphanum,printableascii"`
    ParentID string `bson:"_parent_id" json:"_parent_id" valid:"alphanum,printableascii"`
    Name     string `bson:"name" json:"name"`
    Content  string `bson:"content" json:"content"`
    Slug     string `bson:"slug"`
}

// FindDuplicateEvents returns duplicate Event documents grouped by the specified field
func FindDuplicateEvents(ctx context.Context, coll *mongo.Collection, duplicateField string) ([]bson.M, error) {
    pipeline := []bson.M{
        // Stage 1: Group by target field, count entries, collect document IDs
        {
            "$group": bson.M{
                "_id":         bson.M{duplicateField: "$" + duplicateField},
                "count":       bson.M{"$sum": 1},
                "documentIds": bson.M{"$push": "$_id"},
            },
        },
        // Stage 2: Keep only groups with duplicates (count > 1)
        {
            "$match": bson.M{"count": bson.M{"$gt": 1}},
        },
        // Stage 3: Lookup full documents from the events collection
        {
            "$lookup": bson.M{
                "from":         coll.Name(),
                "localField":   "documentIds",
                "foreignField": "_id",
                "as":           "duplicateDocuments",
            },
        },
        // Optional Stage: Format output for readability
        {
            "$project": bson.M{
                "_id":                0,
                "duplicateFieldValue": "$_id." + duplicateField,
                "duplicateCount":      "$count",
                "duplicateEvents":     "$duplicateDocuments",
            },
        },
    }

    // Run the aggregation
    cursor, err := coll.Aggregate(ctx, pipeline)
    if err != nil {
        return nil, fmt.Errorf("aggregation failed: %w", err)
    }
    defer cursor.Close(ctx)

    // Decode results into a slice of bson.M
    var results []bson.M
    if err := cursor.All(ctx, &results); err != nil {
        return nil, fmt.Errorf("failed to decode results: %w", err)
    }

    return results, nil
}

3. Using the Function #

Here's how to call the function to find duplicates (e.g., by old_id):

func main() {
    // Initialize MongoDB client
    clientOpts := options.Client().ApplyURI("mongodb://localhost:27017")
    client, err := mongo.Connect(context.TODO(), clientOpts)
    if err != nil {
        panic(fmt.Sprintf("failed to connect to MongoDB: %v", err))
    }
    defer func() {
        if err := client.Disconnect(context.TODO()); err != nil {
            panic(fmt.Sprintf("failed to disconnect: %v", err))
        }
    }()

    // Get the events collection
    coll := client.Database("your_database_name").Collection("events")

    // Find duplicates by `old_id`
    duplicates, err := FindDuplicateEvents(context.TODO(), coll, "old_id")
    if err != nil {
        panic(fmt.Sprintf("failed to find duplicates: %v", err))
    }

    // Print and process results
    for _, dup := range duplicates {
        fmt.Printf("Duplicate %s value: %v | Count: %v\n", "old_id", dup["duplicateFieldValue"], dup["duplicateCount"])
        fmt.Println("Duplicate Event details:")
        
        // Decode the bson array into Event structs
        for _, doc := range dup["duplicateEvents"].(bson.A) {
            var event Event
            if bsonBytes, err := bson.Marshal(doc); err == nil {
                bson.Unmarshal(bsonBytes, &event)
                fmt.Printf("- ID: %s | Name: %s | OldID: %s\n", event.ID, event.Name, event.OldID)
            }
        }
        fmt.Println("---")
    }
}

Customization Tips #

• To find duplicates on other fields (like name or slug), just pass the field name (as a string) to FindDuplicateEvents (e.g., FindDuplicateEvents(ctx, coll, "name")).
• If you want to filter duplicates further (e.g., only for events with a specific _parent_id), add an extra $match stage at the start of the pipeline.

内容的提问来源于stack exchange，提问作者fallais