Moteurs de table Executable et ExecutablePool

Les moteurs de table Executable et ExecutablePool vous permettent de définir une table dont les lignes sont générées à partir d’un script que vous définissez (en écrivant des lignes sur stdout). Le script exécutable est stocké dans le répertoire user_scripts et peut lire des données à partir de n’importe quelle source.

• Tables Executable : le script est exécuté à chaque requête
• Tables ExecutablePool : maintiennent un pool de processus persistants et y puisent des processus pour les lectures

Vous pouvez, si vous le souhaitez, inclure une ou plusieurs requêtes d’entrée qui envoient leurs résultats vers stdin pour que le script puisse les lire.

Le moteur de table Executable nécessite deux paramètres : le nom du script et le format des données entrantes. Vous pouvez également transmettre une ou plusieurs requêtes d'entrée :

Executable(script_name, format, [input_query...])

Voici les paramètres pertinents pour une table Executable :

• send_chunk_header
  • Description : envoie le nombre de lignes dans chaque chunk avant d’envoyer un chunk à traiter. Ce paramètre peut vous aider à écrire votre script plus efficacement en préallouant certaines ressources
  • Valeur par défaut : false
• command_termination_timeout
  • Description : délai d’expiration de fin de commande, en secondes
  • Valeur par défaut : 10
• command_read_timeout
  • Description : délai d’expiration pour la lecture des données depuis le stdout de la commande, en millisecondes
  • Valeur par défaut : 10000
• command_write_timeout
  • Description : délai d’expiration pour l’écriture des données vers le stdin de la commande, en millisecondes
  • Valeur par défaut : 10000

Prenons un exemple. Le script Python suivant s’appelle my_script.py et est enregistré dans le dossier user_scripts. Il lit un nombre i et affiche i chaînes aléatoires, chacune précédée d’un nombre séparé par une tabulation :

#!/usr/bin/python3

import sys
import string
import random

def main():

    # Read input value
    for number in sys.stdin:
        i = int(number)

        # Generate some random rows
        for id in range(0, i):
            letters = string.ascii_letters
            random_string = ''.join(random.choices(letters, k=10))
            print(str(id) + '\t' + random_string + '\n', end='')

        # Flush results to stdout
        sys.stdout.flush()

if __name__ == "__main__":
    main()

La table my_executable_table ci-dessous est construite à partir de la sortie de my_script.py, qui génère 10 chaînes aléatoires chaque fois que vous exécutez un SELECT sur my_executable_table :

CREATE TABLE my_executable_table (
   x UInt32,
   y String
)
ENGINE = Executable('my_script.py', TabSeparated, (SELECT 10))

La création de la table est immédiate et n'invoque pas le script. Lorsque vous interrogez my_executable_table, le script est invoqué :

SELECT * FROM my_executable_table

┌─x─┬─y──────────┐
│ 0 │ BsnKBsNGNH │
│ 1 │ mgHfBCUrWM │
│ 2 │ iDQAVhlygr │
│ 3 │ uNGwDuXyCk │
│ 4 │ GcFdQWvoLB │
│ 5 │ UkciuuOTVO │
│ 6 │ HoKeCdHkbs │
│ 7 │ xRvySxqAcR │
│ 8 │ LKbXPHpyDI │
│ 9 │ zxogHTzEVV │
└───┴────────────┘

Les utilisateurs du site Hacker News laissent des commentaires. Python propose une boîte à outils de traitement automatique du langage naturel (nltk) avec un SentimentIntensityAnalyzer permettant de déterminer si les commentaires sont positifs, négatifs ou neutres, notamment en leur attribuant une valeur comprise entre -1 (un commentaire très négatif) et 1 (un commentaire très positif). Créons une table Executable qui calcule le sentiment des commentaires Hacker News à l’aide de nltk.

Cet exemple utilise la table hackernews décrite ici. La table hackernews comprend une colonne id de type UInt64 et une colonne comment de type String. Commençons par définir la table Executable :

CREATE TABLE sentiment (
id UInt64,
sentiment Float32
)
ENGINE = Executable(
    'sentiment.py',
    TabSeparated,
    (SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20)
);

Quelques remarques sur la table sentiment :

• Le fichier sentiment.py est placé dans le dossier user_scripts (le dossier par défaut du paramètre user_scripts_path)
• Le format TabSeparated signifie que notre script Python doit générer des lignes de données brutes contenant des valeurs séparées par des tabulations
• La requête sélectionne deux colonnes de hackernews. Le script Python devra extraire les valeurs de ces colonnes des lignes entrantes

Voici la définition de sentiment.py :

#!/usr/local/bin/python3.9

import sys
import nltk

from nltk.sentiment import SentimentIntensityAnalyzer

def main():
    sentiment_analyzer = SentimentIntensityAnalyzer()

    while True:
        try:
            row = sys.stdin.readline()
            if row == '':
                break

            split_line = row.split("\t")

            id = str(split_line[0])
            comment = split_line[1]

            score = sentiment_analyzer.polarity_scores(comment)['compound']
            print(id + '\t' + str(score) + '\n', end='')
            sys.stdout.flush()
        except BaseException as x:
            break

if __name__ == "__main__":
    main()

Quelques remarques sur notre script Python :

• Pour que cela fonctionne, vous devrez exécuter nltk.downloader.download('vader_lexicon'). Nous aurions pu l’inclure dans le script, mais il serait alors téléchargé chaque fois qu’une requête serait exécutée sur la table sentiment, ce qui n’est pas efficace
• Chaque valeur de row correspond à une ligne du result set de SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20
• La ligne reçue est séparée par des tabulations ; nous extrayons donc id et comment à l’aide de la fonction Python split
• Le résultat de polarity_scores est un objet JSON contenant quelques values. Nous avons choisi de récupérer simplement la value compound de cet objet JSON
• N’oubliez pas que la table sentiment dans ClickHouse utilise le format TabSeparated et contient deux colonnes ; notre fonction print sépare donc ces colonnes par une tabulation

Chaque fois que vous écrivez une requête qui sélectionne des lignes dans la table sentiment, la requête SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20 est exécutée et son résultat est transmis à sentiment.py. Testons cela :

SELECT *
FROM sentiment

La réponse se présente ainsi :

┌───────id─┬─sentiment─┐
│  7398199 │    0.4404 │
│ 21640317 │    0.1779 │
│ 21462000 │         0 │
│ 25168863 │         0 │
│ 25168978 │   -0.1531 │
│ 25169359 │         0 │
│ 25169394 │   -0.9231 │
│ 25169766 │    0.4137 │
│ 25172570 │    0.7469 │
│ 25173687 │    0.6249 │
│ 28291534 │         0 │
│ 28291669 │   -0.4767 │
│ 28291731 │         0 │
│ 28291949 │   -0.4767 │
│ 28292004 │    0.3612 │
│ 28292050 │    -0.296 │
│ 28292322 │         0 │
│ 28295172 │    0.7717 │
│ 28295288 │    0.4404 │
│ 21465723 │   -0.6956 │
└──────────┴───────────┘

La syntaxe de ExecutablePool est similaire à celle de Executable, mais quelques paramètres propres à une table ExecutablePool sont à noter :

• pool_size
  • Description : Taille du pool de processus. Si cette taille est de 0, aucune limite de taille n’est appliquée
  • Valeur par défaut : 16
• max_command_execution_time
  • Description : Durée maximale d’exécution de la commande, en secondes
  • Valeur par défaut : 10

Nous pouvons facilement convertir la table sentiment ci-dessus pour qu’elle utilise ExecutablePool au lieu de Executable :

CREATE TABLE sentiment_pooled (
id UInt64,
sentiment Float32
)
ENGINE = ExecutablePool(
    'sentiment.py',
    TabSeparated,
    (SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20000)
)
SETTINGS
    pool_size = 4;

ClickHouse maintiendra 4 processus à la demande lorsque votre client interrogera la table sentiment_pooled.