Как объединить миллионы строк с помощью EF Core

Question

Я пытаюсь собрать примерно два миллиона строк в зависимости от пользователя. У одного пользователя есть несколько Транзакций, каждая Транзакция имеет платформу и столбцы TransactionType.I, агрегирующие Platform и TransactionType как json, и сохраняются как одна строка.

Но мой код работает медленно. Как я могу улучшить производительность?

  public static void AggregateTransactions()
        {
            using (var db = new ApplicationDbContext())
            {
                db.ChangeTracker.AutoDetectChangesEnabled = false;

                //Get a list of users who have transactions  
                var users = db.Transactions
                   .Select(x => x.User)
                   .Distinct();

                foreach (var user in users.ToList())
                {
                    //Get all transactions for a particular user
                    var _transactions = db.Transactions
                        .Include(x => x.Platform)
                        .Include(x => x.TransactionType)
                        .Where(x => x.User == user)
                        .ToList();

//Aggregate Platforms from all transactions for user
                    Dictionary<string, int> platforms = new Dictionary<string, int>();

                    foreach (var item in _transactions.Select(x => x.Platform).GroupBy(x => x.Name).ToList())
                    {
                        platforms.Add(item.Key, item.Count());
                    };

//Aggregate TransactionTypes from all transactions for user
                   Dictionary<string, int> transactionTypes = new Dictionary<string, int>();

                    foreach (var item in _transactions.Select(x => x.TransactionType).GroupBy(x => x.Name).ToList())
                    {
                        transactionTypes.Add(item.Key, item.Count());
                    };


                    db.Add<TransactionByDay>(new TransactionByDay
                    {
                        User = user,
                        Platforms = platforms,     //The dictionary list is represented as json in table
                        TransactionTypes = transactionTypes     //The dictionary list is represented as json in table
                    });

                    db.SaveChanges();

                }

            }

        }

Обновление

Таким образом, базовый вид данных будет выглядеть следующим образом:

Данные о транзакциях:

Id: b11c6b67-6c74-4bbe-f712-08d609af20cf, UserId: 1, PlatformId: 3, TransactionypeId: 1

Id: 4782803f-2f6b-4d99-f717-08d609af20cf, UserId: 1, PlatformId: 3, TransactionypeId: 4

Агрегировать данные как TransactionPerDay:

Id: 9df41ef2-2fc8-441b-4a2f-08d609e21559, UserId: 1, Платформы: {"p3": 2}, TransactionsTypes: {"t1": 1, "t4": 1}

Таким образом, в этом случае две транзакции объединяются в одну. Вы можете видеть, что платформы и типы транзакций будут агрегированы как json.

Answer 1

Идея состоит в том, чтобы попытаться выполнять меньше запросов и запросов, сначала получая столько данных, сколько необходимо.Таким образом, нет необходимости включать в каждую транзакцию Platform и TransactionType, где вы можете просто запросить их один раз в Dictionary и просмотреть данные.Более того, мы можем выполнить нашу параллельную обработку, а затем сохранить все данные одновременно.

    public static void AggregateTransactions()
    {
        using (var db = new ApplicationDbContext())
        {
            db.ChangeTracker.AutoDetectChangesEnabled = false;

            //Get a list of users who have transactions  
            var transactionsByUser = db.Transactions
               .GroupBy(x => x.User) //Not sure if EF Core supports this kind of grouping
               .ToList();

            var platforms = db.Platforms.ToDictionary(ks => ks.PlatformId);
            var Transactiontypes = db.TransactionTypes.ToDictionary(ks => ks.TransactionTypeId);
            var bag = new ConccurentBag<TransactionByDay>();

            Parallel.ForEach(transactionsByUser, transaction => 
            {
                //Aggregate Platforms from all transactions for user
                Dictionary<string, int> platforms = new Dictionary<string, int>(); //This can be converted to a ConccurentDictionary

                //This can be converted to Parallel.ForEach
                foreach (var item in _transactions.Select(x => platforms[x.PlatformId]).GroupBy(x => x.Name).ToList())
                {
                    platforms.Add(item.Key, item.Count());
                };

               //Aggregate TransactionTypes from all transactions for user
               Dictionary<string, int> transactionTypes = new Dictionary<string, int>(); //This can be converted to a ConccurentDictionary

                //This can be converted to Parallel.ForEach
                foreach (var item in _transactions.Select(x => Transactiontypes[c.TransactionTypeId]).GroupBy(x => x.Name).ToList())
                {
                    transactionTypes.Add(item.Key, item.Count());
                };

                bag.Add(new TransactionByDay
                {
                    User = transaction.Key,
                    Platforms = platforms,     //The dictionary list is represented as json in table
                    TransactionTypes = transactionTypes     //The dictionary list is represented as json in table
                });
            });

            //Before calling this we may need to check the status of the Parallel ForEach, or just convert it back to regular foreach loop if you see no benefit.
            db.AddRange(bag);
            db.SaveChanges();
        }
    }

Вариация # 2

    public static void AggregateTransactions()
    {
        using (var db = new ApplicationDbContext())
        {
            db.ChangeTracker.AutoDetectChangesEnabled = false;

            //Get a list of users who have transactions  
            var users = db.Transactions
               .Select(x => x.User)
               .Distinct().ToList();

            var platforms = db.Platforms.ToDictionary(ks => ks.PlatformId);
            var Transactiontypes = db.TransactionTypes.ToDictionary(ks => ks.TransactionTypeId);
            var bag = new ConccurentBag<TransactionByDay>();

            Parallel.ForEach(users, user => 
            {
                var _transactions = db.Transactions
                .Where(x => x.User == user)
                .ToList();

                //Aggregate Platforms from all transactions for user
                Dictionary<string, int> userPlatforms = new Dictionary<string, int>();
                Dictionary<string, int> userTransactions = new Dictionary<string, int>();

                foreach(var transaction in _transactions)
                {
                   if(platforms.TryGetValue(transaction.PlatformId, out var platform))
                   {
                       if(userPlatforms.TryGetValue(platform.Name, out var tmp))
                       {
                           userPlatforms[platform.Name] = tmp + 1;
                       }
                       else
                       {
                           userPlatforms.Add(platform.Name, 1);
                       }
                   }

                   if(Transactiontypes.TryGetValue(transaction.TransactionTypeId, out var type))
                   {
                       if(userTransactions.TryGetValue(type.Name, out var tmp))
                       {
                           userTransactions[type.Name] = tmp + 1;
                       }
                       else
                       {
                           userTransactions.Add(type.Name, 1);
                       }
                   }
                }

                bag.Add(new TransactionByDay
                {
                    User = user,
                    Platforms = userPlatforms,     //The dictionary list is represented as json in table
                    TransactionTypes = userTransactions     //The dictionary list is represented as json in table
                });

            });

            db.AddRange(bag);
            db.SaveChanges();

        }
    }

Answer 2

Linq To EF не построен для скорости (LinqToSQL проще и быстрее, ИМХО, или вы можете запускать прямые команды SQL с Linq EF \ SQL).Во всяком случае, я не знаю, как это будет скорость:

    using (var db = new MyContext(connectionstring))
    {

        var tbd = (from t in db.Transactions
                    group t by t.User
                    into g
                    let platforms = g.GroupBy(tt => tt.Platform.Name)
                    let trantypes = g.GroupBy(tt => tt.TransactionType.Name)
                    select new {
                       User = g.Key,
                       Platforms = platforms, 
                       TransactionTypes = trantypes 
                    }).ToList()
                    .Select(u => new TransactionByDay {
                        User=u.User, 
                        Platforms=u.Platforms.ToDictionary(tt => tt.Key, tt => tt.Count()),
                        TransactionTypes = u.TransactionTypes.ToDictionary(tt => tt.Key, tt => tt.Count())
                    });
 //...
}

Answer 3

Вы, вероятно, не должны вызывать db.saveChanges () внутри цикла. Помогите поставить его вне цикла, чтобы сохранить изменения один раз.

Но, сказав это, имея дело с большими объемами данных и производительностью, я понял, что ADO.NET, вероятно, является лучшим выбором. Это не означает, что вы должны прекратить использование Entity Framework, но, возможно, для этого метода вы могли бы использовать ADO.NET. Если вы пойдете по этому пути, вы можете либо:

1. Создайте хранимую процедуру, чтобы возвращать данные, с которыми вам нужно работать, заполнять таблицу данных, манипулировать данными и сохранять все в массе, используя sqlBulkCopy.

2. Используйте хранимую процедуру для полного выполнения этой операции. Это исключает необходимость переноса данных в ваше приложение, и вся обработка может происходить в самой базе данных.