Purge DynamoDB table concurrent

purgeDynamoDB.go

package main

import (
	"flag"
	"log"
	"os"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
)

type Item struct {
	Path string `json:"Path"`
	Key  string `json:"Key"`
}

func main() {
	table := flag.String("t", "", "specifies the DynamoDB table name")
	flag.Parse()
	if *table != "" {
		log.Printf("clearing table name: %s\n", *table)
	} else {
		log.Println("no table name defined")
		os.Exit(1)
	}
	sess, _ := session.NewSession()
	// sess, _ := session.NewSession(&aws.Config{Region: aws.String("us-east-1")})
	svc := dynamodb.New(sess)
	params := &dynamodb.ScanInput{TableName: table}

	numItems := 0
	for {
		start := time.Now()

		result, err := svc.Scan(params)
		if err != nil {
			log.Fatalf("Query API call failed: %s", err)
		}

		var wg sync.WaitGroup

		for _, i := range result.Items {
			wg.Add(1)
			go func(i map[string]*dynamodb.AttributeValue, wg *sync.WaitGroup) {
				defer wg.Done()
				item := Item{}
				err = dynamodbattribute.UnmarshalMap(i, &item)
				if err != nil {
					log.Fatalf("Got error unmarshalling: %s", err)
				}

				av, err := dynamodbattribute.MarshalMap(item)
				if err != nil {
					log.Println("Got error marshalling map:", err.Error())
				}
				_, err = svc.DeleteItem(&dynamodb.DeleteItemInput{Key: av, TableName: table})
				if err != nil {
					log.Println("Got error calling DeleteItem", err.Error())
				}
			}(i, &wg)
			numItems++
		}
		wg.Wait()
		elapsed := time.Since(start)
		log.Printf("deleted: %d in %s = %f.0/s", len(result.Items), elapsed, float64(len(result.Items))/elapsed.Seconds())
		if result.LastEvaluatedKey == nil {
			break
		}
	}
	log.Printf("deleted: %d", numItems)
}