Ужасно медленная вставка Azure и удаление пакетных операций

Ответ 1

Хорошо, третий отвечает на обаяние?

http://blogs.msdn.com/b/windowsazurestorage/archive/2010/11/06/how-to-get-most-out-of-windows-azure-tables.aspx

Несколько вещей - эмулятор хранилища - от друга, который сделал некоторые серьезные рывки в нем.

"Все попадает в одну таблицу в одной базе данных (большее количество разделов не влияет ни на что). Каждая операция вставки таблицы составляет не менее 3 sql-операций. Каждая партия находится внутри транзакции. В зависимости от уровня изоляции транзакции эти партии будут иметь ограниченную возможность выполнять параллельно.

Последовательные партии должны быть быстрее, чем отдельные вставки из-за поведения сервера sql. (Отдельные вставки по существу являются небольшими транзакциями, каждый из которых скрывается на диске, а реальная транзакция сбрасывается на диск в виде группы).

IE, использующий несколько разделов, не влияет на производительность на эмуляторе, пока он работает с реальным лазурным хранилищем.

Также включите ведение журнала и немного проверите свои журналы - c:\users\username\appdata\local\developmentstorage

Размер партии 100, по-видимому, обеспечивает лучшую реальную производительность, отключает нагг, выключает ожидание 100, увеличит лимит соединения.

Также убедитесь, что вы случайно не вставляете дубликаты, что приведет к ошибке и замедлит все пути вниз.

и проверить реальное хранилище. Там довольно приличная библиотека, которая обрабатывает большую часть этого для вас - http://www.nuget.org/packages/WindowsAzure.StorageExtensions/, просто убедитесь, что вы на самом деле вызываете ToList на добавлениях и т.д. так как это не будет выполняться до тех пор, пока не будет перечислено. Кроме того, эта библиотека использует динамичность и, следовательно, имеет небольшой перфективный эффект для сериализации, но она позволяет использовать чистые объекты POCO без элементов TableEntity.

~ JT

Ответ 2

базовая концепция - используйте параллелизм, чтобы ускорить это.

шаг 1 - дайте потоку достаточно потока, чтобы снять это - ThreadPool.SetMinThreads(1024, 256);

Шаг 2 - используйте разделы. Я использую guids как идентификаторы, я использую последние для символов, чтобы разделить на 256 уникальных партитантов (на самом деле я группирую их в N подмножеств в моем случае 48 разделов)

шаг 3 - вставка с использованием задач, я использую объединение объектов для таблиц refs

public List<T> InsertOrUpdate(List<T> items)
        {
            var subLists = SplitIntoPartitionedSublists(items);

            var tasks = new List<Task>();

            foreach (var subList in subLists)
            {
                List<T> list = subList;
                var task = Task.Factory.StartNew(() =>
                    {
                        var batchOp = new TableBatchOperation();
                        var tableRef = GetTableRef();

                        foreach (var item in list)
                        {
                            batchOp.Add(TableOperation.InsertOrReplace(item));
                        }

                        tableRef.ExecuteBatch(batchOp);
                        ReleaseTableRef(tableRef);
                    });
                tasks.Add(task);
            }

            Task.WaitAll(tasks.ToArray());

            return items;
        }

private IEnumerable<List<T>> SplitIntoPartitionedSublists(IEnumerable<T> items)
        {
            var itemsByPartion = new Dictionary<string, List<T>>();

            //split items into partitions
            foreach (var item in items)
            {
                var partition = GetPartition(item);
                if (itemsByPartion.ContainsKey(partition) == false)
                {
                    itemsByPartion[partition] = new List<T>();
                }
                item.PartitionKey = partition;
                item.ETag = "*";
                itemsByPartion[partition].Add(item);
            }

            //split into subsets
            var subLists = new List<List<T>>();
            foreach (var partition in itemsByPartion.Keys)
            {
                var partitionItems = itemsByPartion[partition];
                for (int i = 0; i < partitionItems.Count; i += MaxBatch)
                {
                    subLists.Add(partitionItems.Skip(i).Take(MaxBatch).ToList());
                }
            }

            return subLists;
        }

        private void BuildPartitionIndentifiers(int partitonCount)
        {
            var chars = new char[] { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' }.ToList();
            var keys = new List<string>();

            for (int i = 0; i < chars.Count; i++)
            {
                var keyA = chars[i];
                for (int j = 0; j < chars.Count; j++)
                {
                    var keyB = chars[j];
                    keys.Add(string.Concat(keyA, keyB));
                }
            }


            var keySetMaxSize = Math.Max(1, (int)Math.Floor((double)keys.Count / ((double)partitonCount)));
            var keySets = new List<List<string>>();

            if (partitonCount > keys.Count)
            {
                partitonCount = keys.Count;
            }

            //Build the key sets
            var index = 0;
            while (index < keys.Count)
            {
                var keysSet = keys.Skip(index).Take(keySetMaxSize).ToList();
                keySets.Add(keysSet);
                index += keySetMaxSize;
            }

            //build the lookups and datatable for each key set
            _partitions = new List<string>();
            for (int i = 0; i < keySets.Count; i++)
            {
                var partitionName = String.Concat("subSet_", i);
                foreach (var key in keySets[i])
                {
                    _partitionByKey[key] = partitionName;
                }
                _partitions.Add(partitionName);
            }

        }

        private string GetPartition(T item)
        {
            var partKey = item.Id.ToString().Substring(34,2);
            return _partitionByKey[partKey];
        }

        private string GetPartition(Guid id)
        {
            var partKey = id.ToString().Substring(34, 2);
            return _partitionByKey[partKey];
        }

        private CloudTable GetTableRef()
        {
            CloudTable tableRef = null;
            //try to pop a table ref out of the stack
            var foundTableRefInStack = _tableRefs.TryPop(out tableRef);
            if (foundTableRefInStack == false)
            {
                //no table ref available must create a new one
                var client = _account.CreateCloudTableClient();
                client.RetryPolicy = new ExponentialRetry(TimeSpan.FromSeconds(1), 4);
                tableRef = client.GetTableReference(_sTableName);
            }

            //ensure table is created
            if (_bTableCreated != true)
            {
                tableRef.CreateIfNotExists();
                _bTableCreated = true;
            }

            return tableRef;
        }

результат - максимальный размер хранилища 19-22kops

ударил меня, если вы заинтересованы в полном источнике

нужен moar? используйте несколько учетных записей!

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

Ответ 3

После многих испытаний, экспериментов, наконец, удалось получить оптимальную пропускную способность для одного табличного раздела (2000+ операций пакетной записи в секунду) и гораздо более высокую пропускную способность в учетной записи хранилища (3,500+ операций пакетной записи в секунду) с помощью Azure Хранилище таблиц. Я пробовал все разные подходы, но установил лимит подключения .net программно (я попробовал образец конфигурации, но не работал у меня) решил проблему (на основе Белая книга, предоставленная Microsoft), как показано ниже:

ServicePoint tableServicePoint = ServicePointManager
    .FindServicePoint(_StorageAccount.TableEndpoint);

//This is a notorious issue that has affected many developers. By default, the value 
//for the number of .NET HTTP connections is 2.
//This implies that only 2 concurrent connections can be maintained. This manifests itself
//as "underlying connection was closed..." when the number of concurrent requests is
//greater than 2.

tableServicePoint.ConnectionLimit = 1000;

Кто-нибудь, кто получил операцию записи в 20K + на одну учетную запись хранилища, поделитесь своим опытом.

Ответ 4

Для более интересного, здесь новый ответ - изолированный независимый тест, который тянет некоторые удивительные числа для производительности записи на производстве и, черт возьми, намного лучше избегает блокировки IO и управления соединениями. Мне очень интересно посмотреть, как это работает для вас, поскольку мы получаем смешные скорости записи ( > 7 кбит/с).

WebConfig

 <system.net>
    <connectionManagement>
      <add address="*" maxconnection="48"/>
    </connectionManagement>
  </system.net>

Для теста я использовал параметры, основанные на объеме, так что, как и 25000 элементов, 24 раздела, размер пакета из 100, как представляется, всегда являются лучшими и ref count 20. Это использует поток данных TPL (http://www.nuget.org/packages/Microsoft.Tpl.Dataflow/) для BufflerBlock, который обеспечивает хорошее отслеживание ссылки на безопасную таблицу потоков.

public class DyanmicBulkInsertTestPooledRefsAndAsynch : WebTest, IDynamicWebTest
{
    private int _itemCount;
    private int _partitionCount;
    private int _batchSize;
    private List<TestTableEntity> _items;
    private GuidIdPartitionSplitter<TestTableEntity> _partitionSplitter;
    private string _tableName;
    private CloudStorageAccount _account;
    private CloudTableClient _tableClient;
    private Dictionary<string, List<TestTableEntity>> _itemsByParition;
    private int _maxRefCount;
    private BufferBlock<CloudTable> _tableRefs;


    public DyanmicBulkInsertTestPooledRefsAndAsynch()
    {
        Properties = new List<ItemProp>();    
        Properties.Add(new ItemProp("ItemCount", typeof(int)));
        Properties.Add(new ItemProp("PartitionCount", typeof(int)));
        Properties.Add(new ItemProp("BatchSize", typeof(int)));
        Properties.Add(new ItemProp("MaxRefs", typeof(int)));


    }

    public List<ItemProp> Properties { get; set; }

    public void SetProps(Dictionary<string, object> propValuesByPropName)
    {
        _itemCount = (int)propValuesByPropName["ItemCount"];
        _partitionCount = (int)propValuesByPropName["PartitionCount"];
        _batchSize = (int)propValuesByPropName["BatchSize"];
        _maxRefCount = (int)propValuesByPropName["MaxRefs"];
    }

    protected override void SetupTest()
    {
        base.SetupTest();

        ThreadPool.SetMinThreads(1024, 256);
        ServicePointManager.DefaultConnectionLimit = 256;
        ServicePointManager.UseNagleAlgorithm = false;
        ServicePointManager.Expect100Continue = false;


        _account = CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("DataConnectionString"));
        _tableClient = _account.CreateCloudTableClient();
        _tableName = "testtable" + new Random().Next(100000);

        //create the refs
        _tableRefs = new BufferBlock<CloudTable>();
        for (int i = 0; i < _maxRefCount; i++)
        {
            _tableRefs.Post(_tableClient.GetTableReference(_tableName));
        }

        var tableRefTask = GetTableRef();
        tableRefTask.Wait();
        var tableRef = tableRefTask.Result;

        tableRef.CreateIfNotExists();
        ReleaseRef(tableRef);

        _items = TestUtils.GenerateTableItems(_itemCount);
        _partitionSplitter = new GuidIdPartitionSplitter<TestTableEntity>();
        _partitionSplitter.BuildPartitions(_partitionCount);

        _items.ForEach(o =>
            {
                o.ETag = "*";
                o.Timestamp = DateTime.Now;
                o.PartitionKey = _partitionSplitter.GetPartition(o);
            });

        _itemsByParition = _partitionSplitter.SplitIntoPartitionedSublists(_items);
    }

    private async Task<CloudTable> GetTableRef()
    {
        return await _tableRefs.ReceiveAsync();            
    }

    private void ReleaseRef(CloudTable tableRef)
    {
        _tableRefs.Post(tableRef);
    }

    protected override void ExecuteTest()
    {
        Task.WaitAll(_itemsByParition.Keys.Select(parition => Task.Factory.StartNew(() => InsertParitionItems(_itemsByParition[parition]))).ToArray());
    }

    private void InsertParitionItems(List<TestTableEntity> items)
    {

        var tasks = new List<Task>();

        for (int i = 0; i < items.Count; i += _batchSize)
        {
            int i1 = i;

            var task = Task.Factory.StartNew(async () =>
            {
                var batchItems = items.Skip(i1).Take(_batchSize).ToList();

                if (batchItems.Select(o => o.PartitionKey).Distinct().Count() > 1)
                {
                    throw new Exception("Multiple partitions batch");
                }

                var batchOp = new TableBatchOperation();
                batchItems.ForEach(batchOp.InsertOrReplace);   

                var tableRef = GetTableRef.Result();
                tableRef.ExecuteBatch(batchOp);
                ReleaseRef(tableRef);
            });

            tasks.Add(task);

        }

        Task.WaitAll(tasks.ToArray());


    }

    protected override void CleanupTest()
    {
        var tableRefTask = GetTableRef();
        tableRefTask.Wait();
        var tableRef = tableRefTask.Result;
        tableRef.DeleteIfExists();
        ReleaseRef(tableRef);
    }

В настоящее время мы работаем над версией, которая может обрабатывать несколько учетных записей хранилища, чтобы надеяться получить некоторые безумные скорости. Кроме того, мы запускаем их на 8 основных виртуальных машинах для больших наборов данных, но с новым неблокирующим IO он должен работать отлично на ограниченном vm. Удачи!

 public class SimpleGuidIdPartitionSplitter<T> where T : IUniqueId
{
    private ConcurrentDictionary<string, string> _partitionByKey = new ConcurrentDictionary<string, string>();
    private List<string> _partitions;
    private bool _bPartitionsBuilt;

    public SimpleGuidIdPartitionSplitter()
    {

    }

    public void BuildPartitions(int iPartCount)
    {
        BuildPartitionIndentifiers(iPartCount);
    }

    public string GetPartition(T item)
    {
        if (_bPartitionsBuilt == false)
        {
            throw new Exception("Partitions Not Built");
        }

        var partKey = item.Id.ToString().Substring(34, 2);
        return _partitionByKey[partKey];
    }

    public string GetPartition(Guid id)
    {
        if (_bPartitionsBuilt == false)
        {
            throw new Exception("Partitions Not Built");
        }

        var partKey = id.ToString().Substring(34, 2);
        return _partitionByKey[partKey];
    }

    #region Helpers
    private void BuildPartitionIndentifiers(int partitonCount)
    {
        var chars = new char[] { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' }.ToList();
        var keys = new List<string>();

        for (int i = 0; i < chars.Count; i++)
        {
            var keyA = chars[i];
            for (int j = 0; j < chars.Count; j++)
            {
                var keyB = chars[j];
                keys.Add(string.Concat(keyA, keyB));
            }
        }


        var keySetMaxSize = Math.Max(1, (int)Math.Floor((double)keys.Count / ((double)partitonCount)));
        var keySets = new List<List<string>>();

        if (partitonCount > keys.Count)
        {
            partitonCount = keys.Count;
        }

        //Build the key sets
        var index = 0;
        while (index < keys.Count)
        {
            var keysSet = keys.Skip(index).Take(keySetMaxSize).ToList();
            keySets.Add(keysSet);
            index += keySetMaxSize;
        }

        //build the lookups and datatable for each key set
        _partitions = new List<string>();
        for (int i = 0; i < keySets.Count; i++)
        {
            var partitionName = String.Concat("subSet_", i);
            foreach (var key in keySets[i])
            {
                _partitionByKey[key] = partitionName;
            }
            _partitions.Add(partitionName);
        }

        _bPartitionsBuilt = true;
    }
    #endregion
}



internal static List<TestTableEntity> GenerateTableItems(int count)
        {
            var items = new List<TestTableEntity>();
            var random = new Random();

            for (int i = 0; i < count; i++)
            {
                var itemId = Guid.NewGuid();

                items.Add(new TestTableEntity()
                {
                    Id = itemId,
                    TestGuid = Guid.NewGuid(),
                    RowKey = itemId.ToString(),
                    TestBool = true,
                    TestDateTime = DateTime.Now,
                    TestDouble = random.Next() * 1000000,
                    TestInt = random.Next(10000),
                    TestString = Guid.NewGuid().ToString(),
                });
            }

            var dupRowKeys = items.GroupBy(o => o.RowKey).Where(o => o.Count() > 1).Select(o => o.Key).ToList();
            if (dupRowKeys.Count > 0)
            {
                throw  new Exception("Dupicate Row Keys");
            }

            return items;
        }

и еще одна вещь - ваше время и то, как рамочка была затронута, указывают на это http://blogs.msdn.com/b/windowsazurestorage/archive/2013/08/08/net-clients-encountering-port-exhaustion-after-installing-kb2750149-or-kb2805227.aspx