c# - registros - Completamente lento en la inserción de la tabla Azure y eliminación de las operaciones por lotes
optimizar insert masivo sql server (4)
Me estoy encontrando con un gran cuello de botella de rendimiento cuando uso el almacenamiento de tablas Azure. Mi deseo es usar tablas como un tipo de caché, por lo que un proceso largo puede dar como resultado entre cientos y miles de filas de datos. La partición y las claves de fila pueden consultar rápidamente los datos.
La consulta funciona bastante rápido (extremadamente rápido cuando solo se usan claves de partición y fila, un poco más lento, pero aún aceptable cuando también se buscan propiedades para una coincidencia particular).
Sin embargo, tanto la inserción como la eliminación de filas es extremadamente lenta.
Aclaración
Quiero aclarar que incluso insertar un solo lote de 100 elementos lleva varios segundos. Esto no es solo un problema con el rendimiento total de miles de filas. Me está afectando cuando solo inserto 100.
Aquí hay un ejemplo de mi código para hacer una inserción por lotes en mi mesa:
static async Task BatchInsert( CloudTable table, List<ITableEntity> entities )
{
int rowOffset = 0;
while ( rowOffset < entities.Count )
{
Stopwatch sw = Stopwatch.StartNew();
var batch = new TableBatchOperation();
// next batch
var rows = entities.Skip( rowOffset ).Take( 100 ).ToList();
foreach ( var row in rows )
batch.Insert( row );
// submit
await table.ExecuteBatchAsync( batch );
rowOffset += rows.Count;
Trace.TraceInformation( "Elapsed time to batch insert " + rows.Count + " rows: " + sw.Elapsed.ToString( "g" ) );
}
}
Estoy usando operaciones por lotes, y aquí hay una muestra de resultados de depuración:
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Starting asynchronous request to http://127.0.0.1:10002/devstoreaccount1.
Microsoft.WindowsAzure.Storage Verbose: 4 : b08a07da-fceb-4bec-af34-3beaa340239b: StringToSign = POST..multipart/mixed; boundary=batch_6d86d34c-5e0e-4c0c-8135-f9788ae41748.Tue, 30 Jul 2013 18:48:38 GMT./devstoreaccount1/devstoreaccount1/$batch.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Preparing to write request data.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Writing request data.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Waiting for response.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Response received. Status code = 202, Request ID = , Content-MD5 = , ETag = .
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Response headers were processed successfully, proceeding with the rest of the operation.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Processing response body.
Microsoft.WindowsAzure.Storage Information: 3 : b08a07da-fceb-4bec-af34-3beaa340239b: Operation completed successfully.
iisexpress.exe Information: 0 : Elapsed time to batch insert 100 rows: 0:00:00.9351871
Como puede ver, este ejemplo toma casi 1 segundo para insertar 100 filas. El promedio parece ser de aproximadamente .8 segundos en mi máquina de desarrollo (3.4 Ghz quad core).
Esto parece ridículo.
Aquí hay un ejemplo de una operación de eliminación por lotes:
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Starting asynchronous request to http://127.0.0.1:10002/devstoreaccount1.
Microsoft.WindowsAzure.Storage Verbose: 4 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: StringToSign = POST..multipart/mixed; boundary=batch_7e3d229f-f8ac-4aa0-8ce9-ed00cb0ba321.Tue, 30 Jul 2013 18:47:41 GMT./devstoreaccount1/devstoreaccount1/$batch.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Preparing to write request data.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Writing request data.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Waiting for response.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Response received. Status code = 202, Request ID = , Content-MD5 = , ETag = .
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Response headers were processed successfully, proceeding with the rest of the operation.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Processing response body.
Microsoft.WindowsAzure.Storage Information: 3 : 4c271cb5-7463-44b1-b2e5-848b8fb10a93: Operation completed successfully.
iisexpress.exe Information: 0 : Elapsed time to batch delete 100 rows: 0:00:00.6524402
Consistentemente durante .5 segundos.
Ejecuté este implementado en Azure (pequeña instancia) también, y he registrado tiempos de 20 minutos para insertar 28000 filas.
Actualmente estoy usando la versión 2.1 RC de Storage Client Library: MSDN Blog
Debo estar haciendo algo muy mal. ¿Alguna idea?
ACTUALIZAR
Intenté el paralelismo con el efecto neto de una mejora de velocidad general (y 8 procesadores lógicos al máximo), pero apenas 150 insertos de fila por segundo en mi máquina de desarrollo.
En general, no es mejor lo que puedo decir, y tal vez incluso peor cuando se despliega en Azure (pequeña instancia).
Aumenté el grupo de subprocesos y aumenté el número máximo de conexiones HTTP para mi WebRole siguiendo estos consejos .
Todavía siento que me estoy perdiendo algo fundamental que está limitando mis inserciones / eliminaciones a 150 ROPS.
ACTUALIZACIÓN 2
Después de analizar algunos registros de diagnóstico de mi pequeña instancia implementada en Azure (utilizando el nuevo registro integrado en el 2.1 RC Storage Client), tengo un poco más de información.
El primer registro de cliente de almacenamiento para una inserción de lote está en 635109046781264034 ticks:
caf06fca-1857-4875-9923-98979d850df3: Starting synchronous request to https://?.table.core.windows.net/.; TraceSource ''Microsoft.WindowsAzure.Storage'' event
Luego, casi 3 segundos después, veo este registro en 635109046810104314 ticks:
caf06fca-1857-4875-9923-98979d850df3: Preparing to write request data.; TraceSource ''Microsoft.WindowsAzure.Storage'' event
Luego, algunos registros más que toman un total de 0.15 segundos y terminan con este en 635109046811645418 ticks que envuelve el inserto:
caf06fca-1857-4875-9923-98979d850df3: Operation completed successfully.; TraceSource ''Microsoft.WindowsAzure.Storage'' event
No estoy seguro de qué hacer con esto, pero es bastante consistente en todos los registros de inserción de lotes que examiné.
ACTUALIZACIÓN 3
Aquí está el código utilizado para la inserción de lotes en paralelo. En este código, solo para probar, me aseguro de insertar cada lote de 100 en una partición única.
static async Task BatchInsert( CloudTable table, List<ITableEntity> entities )
{
int rowOffset = 0;
var tasks = new List<Task>();
while ( rowOffset < entities.Count )
{
// next batch
var rows = entities.Skip( rowOffset ).Take( 100 ).ToList();
rowOffset += rows.Count;
string partition = "$" + rowOffset.ToString();
var task = Task.Factory.StartNew( () =>
{
Stopwatch sw = Stopwatch.StartNew();
var batch = new TableBatchOperation();
foreach ( var row in rows )
{
row.PartitionKey = row.PartitionKey + partition;
batch.InsertOrReplace( row );
}
// submit
table.ExecuteBatch( batch );
Trace.TraceInformation( "Elapsed time to batch insert " + rows.Count + " rows: " + sw.Elapsed.ToString( "F2" ) );
} );
tasks.Add( task );
}
await Task.WhenAll( tasks );
}
Como se indicó anteriormente, esto ayuda a mejorar el tiempo total para insertar miles de filas, pero cada lote de 100 aún demora varios segundos.
ACTUALIZACIÓN 4
Así que creé un nuevo proyecto de Azure Cloud Service, utilizando VS2012.2, con el rol web como una plantilla de página única (la nueva con el ejemplo de TODO).
Esto viene directamente de la caja, no hay nuevos paquetes NuGet ni nada. Utiliza la biblioteca de cliente de Storage v2 de forma predeterminada, y el EDM y las bibliotecas asociadas v5.2.
Simplemente modifiqué el código de HomeController para que sea el siguiente (usando algunos datos aleatorios para simular las columnas que quiero almacenar en la aplicación real):
public ActionResult Index( string returnUrl )
{
ViewBag.ReturnUrl = returnUrl;
Task.Factory.StartNew( () =>
{
TableTest();
} );
return View();
}
static Random random = new Random();
static double RandomDouble( double maxValue )
{
// the Random class is not thread safe!
lock ( random ) return random.NextDouble() * maxValue;
}
void TableTest()
{
// Retrieve storage account from connection-string
CloudStorageAccount storageAccount = CloudStorageAccount.Parse(
CloudConfigurationManager.GetSetting( "CloudStorageConnectionString" ) );
// create the table client
CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
// retrieve the table
CloudTable table = tableClient.GetTableReference( "test" );
// create it if it doesn''t already exist
if ( table.CreateIfNotExists() )
{
// the container is new and was just created
Trace.TraceInformation( "Created table named " + "test" );
}
Stopwatch sw = Stopwatch.StartNew();
// create a bunch of objects
int count = 28000;
List<DynamicTableEntity> entities = new List<DynamicTableEntity>( count );
for ( int i = 0; i < count; i++ )
{
var row = new DynamicTableEntity()
{
PartitionKey = "filename.txt",
RowKey = string.Format( "$item{0:D10}", i ),
};
row.Properties.Add( "Name", EntityProperty.GeneratePropertyForString( i.ToString() ) );
row.Properties.Add( "Data", EntityProperty.GeneratePropertyForString( string.Format( "data{0}", i ) ) );
row.Properties.Add( "Value1", EntityProperty.GeneratePropertyForDouble( RandomDouble( 10000 ) ) );
row.Properties.Add( "Value2", EntityProperty.GeneratePropertyForDouble( RandomDouble( 10000 ) ) );
row.Properties.Add( "Value3", EntityProperty.GeneratePropertyForDouble( RandomDouble( 1000 ) ) );
row.Properties.Add( "Value4", EntityProperty.GeneratePropertyForDouble( RandomDouble( 90 ) ) );
row.Properties.Add( "Value5", EntityProperty.GeneratePropertyForDouble( RandomDouble( 180 ) ) );
row.Properties.Add( "Value6", EntityProperty.GeneratePropertyForDouble( RandomDouble( 1000 ) ) );
entities.Add( row );
}
Trace.TraceInformation( "Elapsed time to create record rows: " + sw.Elapsed.ToString() );
sw = Stopwatch.StartNew();
Trace.TraceInformation( "Inserting rows" );
// batch our inserts (100 max)
BatchInsert( table, entities ).Wait();
Trace.TraceInformation( "Successfully inserted " + entities.Count + " rows into table " + table.Name );
Trace.TraceInformation( "Elapsed time: " + sw.Elapsed.ToString() );
Trace.TraceInformation( "Done" );
}
static async Task BatchInsert( CloudTable table, List<DynamicTableEntity> entities )
{
int rowOffset = 0;
var tasks = new List<Task>();
while ( rowOffset < entities.Count )
{
// next batch
var rows = entities.Skip( rowOffset ).Take( 100 ).ToList();
rowOffset += rows.Count;
string partition = "$" + rowOffset.ToString();
var task = Task.Factory.StartNew( () =>
{
var batch = new TableBatchOperation();
foreach ( var row in rows )
{
row.PartitionKey = row.PartitionKey + partition;
batch.InsertOrReplace( row );
}
// submit
table.ExecuteBatch( batch );
Trace.TraceInformation( "Inserted batch for partition " + partition );
} );
tasks.Add( task );
}
await Task.WhenAll( tasks );
}
Y esta es la salida que obtengo:
iisexpress.exe Information: 0 : Elapsed time to create record rows: 00:00:00.0719448
iisexpress.exe Information: 0 : Inserting rows
iisexpress.exe Information: 0 : Inserted batch for partition $100
...
iisexpress.exe Information: 0 : Successfully inserted 28000 rows into table test
iisexpress.exe Information: 0 : Elapsed time: 00:01:07.1398928
Esto es un poco más rápido que en mi otra aplicación, a más de 460 ROPS. Esto aún es inaceptable. Y de nuevo en esta prueba, mi CPU (8 procesadores lógicos) está casi al máximo, y el acceso al disco está casi inactivo.
Estoy perdido en lo que está mal.
ACTUALIZACIÓN 5
Las vueltas y vueltas de violín y ajustes han producido algunas mejoras, pero simplemente no puedo obtenerlo mucho más rápido que 500-700 (ish) ROPS realizando operaciones por lotes InsertOrReplace (en lotes de 100).
Esta prueba se realiza en la nube de Azure, usando una pequeña instancia (o dos). Basado en los comentarios a continuación, me resigno al hecho de que las pruebas locales serán lentas en el mejor de los casos.
Aquí hay un par de ejemplos. Cada ejemplo es su propia PartitionKey:
Successfully inserted 904 rows into table org1; TraceSource ''w3wp.exe'' event
Elapsed time: 00:00:01.3401031; TraceSource ''w3wp.exe'' event
Successfully inserted 4130 rows into table org1; TraceSource ''w3wp.exe'' event
Elapsed time: 00:00:07.3522871; TraceSource ''w3wp.exe'' event
Successfully inserted 28020 rows into table org1; TraceSource ''w3wp.exe'' event
Elapsed time: 00:00:51.9319217; TraceSource ''w3wp.exe'' event
Tal vez es mi cuenta de MSDN Azure que tiene algunos topes de rendimiento? No lo sé.
En este punto, creo que he terminado con esto. Tal vez sea lo suficientemente rápido para usarlo para mis propósitos, o tal vez siga un camino diferente.
CONCLUSIÓN
Todas las respuestas a continuación son buenas!
Para mi pregunta específica, he podido ver velocidades de hasta 2k ROPS en una pequeña instancia de Azure, más típicamente alrededor de 1k. Como necesito mantener los costos bajos (y por lo tanto los tamaños de las instancias), esto define para lo que podré usar las tablas.
Gracias a todos por toda la ayuda.
Después de experimentar mucho dolor, los experimentos finalmente lograron un rendimiento óptimo para la partición de tabla única (más de 2,000 operaciones de escritura por lotes por segundo) y mucho mejor rendimiento en la cuenta de almacenamiento (más de 3,500 operaciones de escritura por lotes por segundo) con Azure Table Storage. Probé todos los enfoques diferentes, pero establecer el límite de conexión .net programáticamente (probé la configuración de muestra, pero no funcionó) solucionó el problema (basado en un documento técnico proporcionado por Microsoft), como se muestra a continuación:
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;
Alguien más que obtuvo 20K + operación de escritura por lotes por cuenta de almacenamiento, por favor comparta su experiencia.
Ok, el 3º responde con un hechizo?
Un par de cosas, el emulador de almacenamiento, de un amigo que lo investigó seriamente.
"Todo está golpeando una sola tabla en una única base de datos (más particiones no afectan nada). Cada operación de inserción de tabla es de al menos 3 operaciones sql. Cada lote está dentro de una transacción. Dependiendo del nivel de aislamiento de transacción, esos lotes tendrán limitaciones capacidad de ejecutar en paralelo.
Los lotes en serie deberían ser más rápidos que los insertos individuales debido al comportamiento del servidor sql. (Las inserciones individuales son básicamente pequeñas transacciones que cada una se va al disco, mientras que una transacción real se vacía en el disco como un grupo) ".
El uso de varias particiones no afecta el rendimiento en el emulador, mientras que sí lo hace en comparación con el almacenamiento azul real.
También habilite el registro y verifique sus registros un poco - c: / users / username / appdata / local / developmentstorage
El tamaño del lote de 100 parece ofrecer el mejor rendimiento real, desactivar el regateo, desactivar 100, aumentar el límite de conexión.
Asegúrese también de que no está insertando duplicados accidentalmente, eso causará un error y se ralentizará hasta muy abajo.
y prueba contra almacenamiento real. Hay una biblioteca bastante decente que maneja la mayor parte de esto para usted - http://www.nuget.org/packages/WindowsAzure.StorageExtensions/ , solo asegúrese de llamar a ToList en las adiciones y de que realmente no lo hará ejecutar hasta enumerado. Además, esa biblioteca usa dynamictableentity y, por lo tanto, hay un pequeño golpe de percusión para la serialización, pero le permite usar objetos POCO puros sin elementos de TableEntity.
~ JT
Para más diversión, he aquí una nueva respuesta: prueba independiente aislada que está sacando algunos números asombrosos para el rendimiento de escritura en producción y lo hace muchísimo mejor evitando el bloqueo de IO y la administración de la conexión. Estoy muy interesado en ver cómo funciona esto para ti ya que estamos obteniendo velocidades de escritura ridículas (> 7kps).
webconfig
<system.net>
<connectionManagement>
<add address="*" maxconnection="48"/>
</connectionManagement>
</system.net>
Para la prueba, estaba usando parámetros basados en el volumen, por lo que, al igual que 25000 elementos, 24 particiones, el tamaño de lote de 100 parece ser siempre el mejor, y el recuento de ref. 20. Esto es mediante el flujo de datos TPL ( http://www.nuget.org/packages/Microsoft.Tpl.Dataflow/ ) para BufflerBlock, que proporciona una extracción segura y segura de referencias de tablas.
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);
}
Actualmente estamos trabajando en una versión que puede manejar múltiples cuentas de almacenamiento con la esperanza de obtener algunas velocidades insanas. Además, estamos ejecutando estos en máquinas virtuales de 8 núcleos para grandes conjuntos de datos, pero con el nuevo IO no bloqueante debería funcionar muy bien en una vm limitada. ¡Buena suerte!
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;
}
y una cosa más: su tiempo y cómo se vio afectado el marco de trabajo señalan a este http://blogs.msdn.com/b/windowsazurestorage/archive/2013/08/08/net-clients-encountering-port-exhaustion-after-installing-kb2750149-or-kb2805227.aspx
concepto básico: use el paralelismo para acelerar esto.
paso 1: dele a su subproceso suficientes hilos para llevarlo a cabo: ThreadPool.SetMinThreads (1024, 256);
paso 2 - usa particiones. Utilizo guids como Id, utilizo el último para separar los personajes en 256 particiones únicas (de hecho, los agrupo en subconjuntos N en mi caso 48 particiones)
paso 3 - inserte usando tareas, utilizo la agrupación de objetos para referencias de tabla
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;
}
resultado: 19-22kops de cuenta de almacenamiento como máximo
golpeame si estás interesado en la fuente completa
¿Necesitas moar? ¡usa múltiples cuentas de almacenamiento!
esto es de meses de prueba y error, pruebas, golpeando mi cabeza contra un escritorio. Realmente espero que ayude