Motores de tabela Executable e ExecutablePool

Os motores de tabela Executable e ExecutablePool permitem definir uma tabela cujas linhas são geradas por um script que você define (ao gravar linhas em stdout). O script executável é armazenado no diretório user_scripts e pode ler dados de qualquer origem.

• Tabelas Executable: o script é executado em cada consulta
• Tabelas ExecutablePool: mantêm um pool de processos persistentes e usam processos desse pool para leituras

Opcionalmente, é possível incluir uma ou mais consultas de entrada que transmitem seus resultados para stdin, para que o script possa lê-los.

O mecanismo de tabela Executable requer dois parâmetros: o nome do script e o formato dos dados de entrada. Opcionalmente, é possível passar uma ou mais consultas de entrada:

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

Aqui estão as configurações relevantes para uma tabela Executable:

• send_chunk_header
  • Descrição: Envia o número de linhas em cada fragmento antes de enviá-lo para processamento. Essa configuração pode ajudar você a escrever o script de forma mais eficiente, pré-alocando alguns recursos
  • Valor padrão: false
• command_termination_timeout
  • Descrição: Tempo limite para encerramento do comando, em segundos
  • Valor padrão: 10
• command_read_timeout
  • Descrição: Tempo limite para ler dados da stdout do comando, em milissegundos
  • Valor padrão: 10000
• command_write_timeout
  • Descrição: Tempo limite para gravar dados na stdin do comando, em milissegundos
  • Valor padrão: 10000

Vamos ver um exemplo. O script Python a seguir se chama my_script.py e está salvo na pasta user_scripts. Ele lê um número i e imprime i strings aleatórias, cada uma precedida por um número separado por uma tabulação:

#!/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()

A tabela my_executable_table a seguir é criada a partir da saída de my_script.py, que gera 10 strings aleatórias sempre que você executa um SELECT em my_executable_table:

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

A criação da tabela é concluída imediatamente e não executa o script. Ao consultar my_executable_table, o script é executado:

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 │
└───┴────────────┘

Os usuários do site Hacker News deixam comentários. O Python inclui um kit de ferramentas de processamento de linguagem natural (nltk) com um SentimentIntensityAnalyzer para determinar se os comentários são positivos, negativos ou neutros, inclusive atribuindo um valor entre -1 (um comentário muito negativo) e 1 (um comentário muito positivo). Vamos criar uma tabela Executable que calcula o sentimento dos comentários do Hacker News usando nltk.

Este exemplo usa a tabela hackernews descrita aqui. A tabela hackernews inclui uma coluna id do tipo UInt64 e uma coluna String chamada comment. Vamos começar definindo a tabela 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)
);

Alguns comentários sobre a tabela sentiment:

• O arquivo sentiment.py é salvo na pasta user_scripts (a pasta padrão da configuração user_scripts_path)
• O formato TabSeparated significa que nosso script Python precisa gerar linhas de dados brutos com valores separados por tabulação
• A consulta seleciona duas colunas de hackernews. O script Python precisará extrair os valores dessas colunas das linhas recebidas

Aqui está a definição 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()

Alguns comentários sobre nosso script em Python:

• Para que isso funcione, você precisará executar nltk.downloader.download('vader_lexicon'). Isso poderia ter sido incluído no script, mas, nesse caso, seria baixado toda vez que uma consulta fosse executada na tabela sentiment — o que não é eficiente
• Cada valor de row corresponderá a uma linha no conjunto de resultados de SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20
• A linha recebida é separada por tabulação, então extraímos o id e o comment usando a função split do Python
• O resultado de polarity_scores é um objeto JSON com alguns valores. Decidimos simplesmente usar o valor compound desse objeto JSON
• Lembre-se de que a tabela sentiment no ClickHouse usa o formato TabSeparated e contém duas colunas, então nossa função print separa essas colunas com uma tabulação

Toda vez que você escreve uma consulta que seleciona linhas da tabela sentiment, a consulta SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20 é executada, e o resultado é passado para sentiment.py. Vamos testar:

SELECT *
FROM sentiment

A resposta é assim:

┌───────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 │
└──────────┴───────────┘

A sintaxe de ExecutablePool é semelhante à de Executable, mas há algumas configurações importantes exclusivas de uma tabela ExecutablePool:

• pool_size
  • Descrição: Tamanho do pool de processos. Se o tamanho for 0, não haverá restrições de tamanho
  • Valor padrão: 16
• max_command_execution_time
  • Descrição: Tempo máximo de execução do comando, em segundos
  • Valor padrão: 10

Podemos converter facilmente a tabela sentiment acima para usar ExecutablePool em vez 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;

O ClickHouse manterá 4 processos conforme necessário quando seu cliente consultar a tabela sentiment_pooled.