Si l'ordre des résultats de la requête n'est pas important, vous pouvez utiliser la fonctionnalité d'analyse parallèle pour obtenir les résultats de manière efficace.
Prérequis
Une instance OTSClient est initialisée. Pour plus d'informations, consultez Initialiser une instance OTSClient.
Une table de données est créée et des données y sont écrites. Pour plus d'informations, consultez Créer des tables de données et Écrire des données.
Un index de recherche est créé pour la table de données. Pour plus d'informations, consultez Créer des index de recherche.
Paramètres
Une analyse parallèle nécessite deux opérations de coordination : appelez ComputeSplits pour obtenir l'ID de session et le nombre maximal de tâches parallèles (MaxParallel), puis lancez une tâche ParallelScan par emplacement parallèle — chacune avec un CurrentParallelId unique allant de 0 à MaxParallel - 1. Par exemple, si MaxParallel est égal à 4, démarrez quatre tâches d'analyse avec des valeurs CurrentParallelId de 0, 1, 2 et 3 pour couvrir l'ensemble complet des données.
|
Paramètre |
Description |
|
|
TableName |
Le nom de la table de données. |
|
|
IndexName |
Le nom de l'index de recherche. |
|
|
ScanQuery |
Query |
La condition de requête. Types pris en charge : requête term, requête floue, requête de plage, requête géographique et requête imbriquée — les mêmes types que ceux pris en charge par l'opération Search. |
|
Limit |
Le nombre maximal de lignes renvoyées par appel ParallelScan. |
|
|
MaxParallel |
Le nombre maximal de tâches d'analyse parallèle par requête. La valeur maximale dépend du volume de données : les ensembles de données plus volumineux supportent davantage de tâches parallèles. Appelez ComputeSplits pour obtenir cette valeur avant de démarrer une analyse. Chaque requête ParallelScan utilise un CurrentParallelId dans la plage [0, MaxParallel), et tous les ID de cette plage doivent être couverts pour analyser l'ensemble complet des données. Par exemple, si MaxParallel est égal à 4, démarrez quatre tâches avec des valeurs CurrentParallelId de 0, 1, 2 et 3. MaxParallel et CurrentParallelId doivent être utilisés conjointement. |
|
|
CurrentParallelId |
L'ID de cette tâche d'analyse parallèle. Valeurs valides : [0, MaxParallel). Chaque tâche concurrente doit utiliser un ID unique. Doit être utilisé conjointement avec MaxParallel. |
|
|
Token |
Le jeton de pagination pour récupérer la page de résultats suivante. Chaque réponse ParallelScan inclut un jeton pour la page suivante. Transmettez-le dans la requête suivante pour poursuivre la lecture. Lorsque le jeton est nul, toutes les données de cette tâche ont été récupérées. |
|
|
AliveTime |
La période de validité de la session et de son jeton associé, en secondes. Valeur par défaut : 60. Utilisez la valeur par défaut. Si aucune requête n'est envoyée pendant la période de validité, la session expire et aucune autre donnée ne peut être récupérée. La période de validité est réinitialisée à chaque requête. Remarque
Les sessions peuvent expirer prématurément si le schéma de l'index de recherche est modifié dynamiquement, si un seul serveur tombe en panne ou si un équilibrage de charge des serveurs est effectué. Dans ces cas, recréez la session. |
|
|
ColumnsToGet |
Les colonnes à renvoyer. Définissez le paramètre Columns pour spécifier les noms des colonnes. Pour renvoyer toutes les colonnes de l'index de recherche, définissez ReturnAllFromIndex sur true. Important
Le paramètre ReturnAll n'est pas pris en charge. |
|
|
SessionId |
L'ID de session pour l'analyse parallèle. Appelez ComputeSplits pour créer une session et obtenir à la fois l'ID de session et le nombre maximal de tâches parallèles prises en charge (MaxParallel). |
|
Exemple
L'exemple suivant analyse toutes les données à l'aide d'un seul thread. Il appelle ComputeSplits pour obtenir l'ID de session et MaxParallel, puis parcourt les résultats à l'aide de NextToken jusqu'à ce que toutes les lignes soient récupérées.
/// <summary>
/// Scans all data in a search index using a single thread.
/// Calls ComputeSplits to get the session ID and MaxParallel,
/// then pages through results until NextToken is null.
/// </summary>
public class ParallelScan
{
public static void ParallelScanwithSingleThread(OTSClient otsClient)
{
SearchIndexSplitsOptions options = new SearchIndexSplitsOptions
{
IndexName = IndexName
};
ComputeSplitsRequest computeSplitsRequest = new ComputeSplitsRequest
{
TableName = TableName,
SplitOptions = options
};
ComputeSplitsResponse computeSplitsResponse = otsClient.ComputeSplits(computeSplitsRequest);
MatchAllQuery matchAllQuery = new MatchAllQuery();
ScanQuery scanQuery = new ScanQuery();
scanQuery.AliveTime = 60;
scanQuery.Query = matchAllQuery;
scanQuery.MaxParallel = computeSplitsResponse.SplitsSize;
scanQuery.Limit = 10;
ParallelScanRequest parallelScanRequest = new ParallelScanRequest();
parallelScanRequest.TableName = TableName;
parallelScanRequest.IndexName = IndexName;
parallelScanRequest.ScanQuery = scanQuery;
parallelScanRequest.ColumnToGet = new ColumnsToGet { ReturnAllFromIndex = true };
parallelScanRequest.SessionId = computeSplitsResponse.SessionId;
int total = 0;
List<Row> result = new List<Row>();
ParallelScanResponse parallelScanResponse = otsClient.ParallelScan(parallelScanRequest);
while (parallelScanResponse.NextToken != null)
{
List<Row> rows = new List<Row>(parallelScanResponse.Rows);
total += rows.Count;
result.AddRange(rows);
parallelScanRequest.ScanQuery.Token = parallelScanResponse.NextToken;
parallelScanResponse = otsClient.ParallelScan(parallelScanRequest);
}
foreach (Row row in result)
{
Console.WriteLine(JsonConvert.SerializeObject(row));
}
Console.WriteLine("Total Row Count: {0}", total);
}
}