Récupération de données avec pagination et curseurs

Informations sur les données à grande échelle

En tant que système de base de données distribué prenant en charge le langage de requête SQL standard, CrateDB est idéal pour exécuter des agrégations côté serveur, travailler sur d’énormes ensembles de données et obtenir des résultats résumés ; il y a cependant des cas où l’on peut encore vouloir récupérer beaucoup de données de CrateDB, pour entraîner un modèle d’apprentissage automatique par exemple.

CrateDB collecte les résultats en mémoire avant de les renvoyer aux clients, essayant donc d’exécuter un SELECT une instruction qui renvoie un ensemble de résultats très volumineux en une seule fois peut déclencher disjoncteurs ou entraîner une OutOfMemoryErroret obtenir tous les résultats en une seule opération peut également être un défi côté client, nous avons donc besoin d’un mécanisme pour récupérer les résultats dans des lots gérables lorsque nous sommes prêts à les traiter.

Une option, pour les cas où nous examinons une seule table, et nous savons déjà que nous avons besoin de tous les enregistrements qui satisfont à une condition, est de faire une exportation en bloc avec le COPIER commande qui accepte un WHERE clause. Il arrive cependant que, dans de nombreux cas, nous souhaitions exécuter des requêtes plus complexes, ou que le simple stockage des résultats dans des fichiers ne s’intègre pas bien dans notre application.

Une exigence courante est également ce qu’on appelle la pagination, qui consiste à présenter aux utilisateurs des résultats avec des pages contenant chacune un nombre défini de résultats, leur permettant de se déplacer entre ces pages. Dans ce cas, il est courant que de nombreux utilisateurs ne consultent que les premières pages de résultats, nous souhaitons donc mettre en œuvre cela de la manière la plus efficace possible.

Imaginons que nous ayons un tableau appelé « observations » avec les données suivantes :

+---------------------+--------+---------+
|         ts          | device | reading |
+---------------------+--------+---------+
| 2021-10-14T09:39:19 | dev2   |   -1682 |
| 2022-02-02T00:33:47 | dev1   |     827 |
| 2022-06-11T21:49:53 | dev2   |    -713 |
| 2022-06-29T23:23:28 | dev1   |    1059 |
| 2022-07-01T09:22:56 | dev2   |    -689 |
| 2022-07-10T02:43:36 | dev2   |    -570 |
| 2022-09-18T03:28:02 | dev1   |     303 |
| 2022-10-14T20:34:10 | dev1   |    1901 |
+---------------------+--------+---------+

Nous travaillerons ici avec un très petit nombre d’enregistrements pour visualiser le fonctionnement des différentes techniques mais imaginons que nous ayons des milliers voire des millions d’enregistrements.

En particulier, je montrerai ici des exemples de récupération de 2 lignes à la fois, mais selon le cas d’utilisation, vous chercherez probablement à récupérer 50, 1 000 ou même 5 000 lignes à la fois.

SELECT date_format(ts),device,reading 
FROM doc.observations 
WHERE ts BETWEEN '2022-01-01 00:00' AND '2022-10-01 00:00'
ORDER by ts
LIMIT 2;

Cela renvoie :

+-----------------------------+--------+---------+
| date_format(ts)             | device | reading |
+-----------------------------+--------+---------+
| 2022-02-02T00:33:47.000000Z | dev1   |     827 |
| 2022-06-11T21:49:53.000000Z | dev2   |    -713 |
+-----------------------------+--------+---------+

Nous pourrions alors relancer la requête avec LIMIT 2 OFFSET 2 et on obtiendrait :

+-----------------------------+--------+---------+
| date_format(ts)             | device | reading |
+-----------------------------+--------+---------+
| 2022-06-29T23:23:28.000000Z | dev1   |    1059 |
| 2022-07-01T09:22:56.000000Z | dev2   |    -689 |
+-----------------------------+--------+---------+

Il y a un certain nombre de problèmes à prendre en compte avec cette approche.

Chaque nouvelle requête est considérée comme une nouvelle requête et examine les données actuelles. Considérez ce qui se passe si l’observation du 11 juin 2022 ci-dessus devait être supprimée après l’exécution de la première requête, mais avant d’exécuter la seconde avec OFFSET 2. En sautant 2 lignes, nous ignorons maintenant l’observation du 29 juin 2022, et les utilisateurs ne la verront jamais.

Un autre problème est qu’il n’y a pas toujours un moyen efficace pour CrateDB d’ignorer les lignes, donc, pour certaines requêtes, comme le OFFSET augmente, nous pouvons constater que les temps d’exécution augmentent de plus en plus à mesure que le moteur parcourt les lignes qui doivent être ignorées et les supprime simplement côté serveur.

En continuant de l’exemple ci-dessus, après avoir obtenu les 2 premières lignes, au lieu d’utiliser OFFSET 2 nous pourrions exécuter une requête comme celle-ci :

SELECT date_format(ts),device,reading 
FROM doc.observations 
WHERE ts > '2022-06-11T21:49:53.000000Z' AND ts <='2022-10-01 00:00'
ORDER by ts
LIMIT 2;

Cette valeur du 11 juin est la dernière valeur que nous avons observée jusqu’à présent sur la colonne ts que dans ce cas, nous savons qu’elle augmente toujours, cette approche est très efficace, mais elle ne peut être utilisée que s’il existe un champ approprié dans les données qui n’est pas toujours le cas.

Aussi par rapport à la LIMIT + OFFSET approche dont nous avons parlé précédemment, nous ne pouvons pas l’utiliser pour permettre aux utilisateurs d’accéder à une page de résultats donnée sans avoir d’abord obtenu tous les résultats des pages précédentes, nous ne pouvons par exemple pas accéder directement à la page 10 car nous ne savons pas quelle est la dernière lecture de ts à la page 9.

Certaines personnes appellent cette approche ci-dessus « pagination du curseur », mais le concept le plus courant derrière les « curseurs » est quelque chose d’un peu différent dont nous allons discuter maintenant.

Un curseur est comme avoir un signet pointant vers un enregistrement spécifique dans le jeu de résultats d’une requête, c’est une approche générique qui est implémentée efficacement et ne nous oblige pas à avoir une colonne spéciale ancre/filigrane.

Dans CrateDB, nous pouvons utiliser des curseurs au niveau du protocole ou avec des commandes SQL.

Les curseurs dans CrateDB sont INSENSITIVEce qui signifie que le client peut prendre tout le temps dont il a besoin pour récupérer les résultats, et les données refléteront toujours l’état des tables tel qu’il était au moment où le curseur a été déclaré, ignorant tous les enregistrements qui ont été mis à jour, supprimés ou nouvellement inséré.

En Python, une façon de travailler avec les curseurs est avec asyncpgprofitant de la compatibilité de CrateDB avec le Protocole filaire PostgreSQL.
Tout d’abord, nous devons installer la bibliothèque :

pip install asyncpg

Ensuite, nous pouvons l’utiliser comme ceci:

import asyncio
import asyncpg
# If you are using jupyter-notebook 
# remove this function definition line and the indentation in the block of code that follows
async def main():
# we will then establish a connection    
conn = await asyncpg.connect(host='localhost', user='crate')
# and we need a "transaction" context, 
# there are no transactions as such in CrateDB, 
# but this gives a scope where the cursor lives:  
async with conn.transaction():
# and now we can declare the cursor 
# specifying how many rows we want asyncpg to fetch at a time from CrateDB, 
# and we can iterate over the results: 
query = "SELECT ts,device,reading FROM doc.observations WHERE ts BETWEEN '2022-01-01 00:00' AND '2022-10-01 00:00';"
async for record in conn.cursor(query, prefetch=1000):
print(record)
# Remove this line if you are using jupyter-notebook
asyncio.run(main())

Juste pour clarifier, notre code Python fonctionne avec un enregistrement à la fois, mais dans les coulisses asyncpg demande 1000 enregistrements à la fois à CrateDB.

En Java, nous pouvons utiliser le pilote PostgreSQL JDBC.
Dans un projet Maven, ajoutez ceci à votre pom.xml:

<dependencies>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<version>42.5.0</version>
</dependency>
</dependencies>

Ensuite, nous pouvons l’utiliser comme ceci:

import java.sql.*;
/* ... */	
/* we first establish a connection to CrateDB */
try (Connection conn = DriverManager.getConnection("jdbc:postgresql://localhost/", "crate","")) {
try (Statement st = conn.createStatement()) {
/* We will then open the cursor 
* defining how many rows we want to retrieve at a time, 
* in this case 1,000: 
*/
st.setFetchSize(1000);
String query = "SELECT ts,device,reading ";
query += "FROM doc.observations ";
query += "WHERE ts BETWEEN '2022-01-01 00:00' AND '2022-10-01 00:00';"; 
try (ResultSet resultSet = st.executeQuery(query)) {
/* and while there are rows available, we will iterate over them: */
while (resultSet.next()) {		            	
System.out.println(resultSet.getDate("ts").toString());		            	
}
}
}
}

Cela fonctionne comme dans le cas Python ci-dessus, dans notre code Java, nous voyons une ligne à la fois, mais les lignes sont extraites de CrateDB 1 000 à la fois et conservées en mémoire côté client.

Une approche qui fonctionne avec tous les clients consiste à utiliser Commandes SQL pris en charge depuis CrateDB 5.1.0.

Tout d’abord, nous devons émettre cette commande :

BEGIN;

Il s’agit d’une commande SQL qui démarrerait normalement une transaction, il n’y a pas de transaction en tant que telle dans CrateDB, mais cela créera une portée sur laquelle des curseurs peuvent être créés.

DECLARE observations_cursor NO SCROLL CURSOR FOR 
SELECT ts,device,reading 
FROM doc.observations 
WHERE ts BETWEEN '2022-01-01 00:00' AND '2022-10-01 00:00';

Cela associe un nom de curseur à une requête et détermine le moment auquel les données sont « figées » du point de vue du curseur.

FETCH 10 FROM observations_cursor;

Cela récupère 10 lignes de la requête, et lorsqu’il est émis à nouveau, il récupère les 10 lignes suivantes et ainsi de suite. Nous pouvons récupérer un nombre différent d’enregistrements à chaque fois et nous savons que nous avons atteint la fin du jeu de résultats lorsque FETCH renvoie zéro ligne.

Une fois que le curseur n’est plus nécessaire, il peut être fermé soit avec END;, CLOSE ALL;, CLOSE observations_cursor;, COMMIT;, COMMIT TRANSACTION;ou alors COMMIT WORK;.

Jetez un œil à cette courte animation montrant un exemple de la façon dont cela fonctionne :