A equipe Amazon Kinesis liberou, em outubro de 2014, a Kinesis Client Library (KCL) para desenvolvedores Python. Os desenvolvedores podem usar a KCL para construir aplicações distribuídas que processam streaming de dados de forma confiável em escala. O KCL cuida de muitas das tarefas complexas associadas à computação distribuída, tais como o balanceamento de carga em várias instâncias – em resposta às falhas de instância – e registros processados de pontos de verificação (checkpointing) e reação a mudanças no stream volume.

Faça o download do KCL para Python usando Github ou PyPi.

Primeiros passos

Uma vez que você esteja familiarizado com os conceitos principais do Kinesis e do KCL, estará pronto para escrever sua primeira aplicação. Seu código possui as seguintes atribuições:

1. Configurar os parâmetros de configuração da aplicação
2. Implementar um processador de registro

Os parâmetros de configuração da aplicação são especificados pela adição de um arquivo de propriedades. Por exemplo:

# The python executable script 
executableName = sample_kclpy_app.py
view sourceprint?

# The name of an Amazon Kinesis stream to process.
streamName = words3
 
# Unique KCL application name
applicationName = PythonKCLSample

# Read from the beginning of the stream
initialPositionInStream = TRIM_HORIZON

O exemplo acima configura o KCL para processar um fluxo Kinesis chamado “words” (palavras) usando o processador de registro fornecido em sample_kclpy_app.py. O nome exclusivo da aplicação é usado para dar as coordenadas entre tarefas executando em várias instâncias.

Os desenvolvedores têm que implementar os três métodos a seguir em seus processadores de registro:

initialize(self, shard_id)
process_records(self, records, checkpointer)
shutdown(self, checkpointer, reason)

O initialize() e shutdown()são auto explicativos; eles são chamados uma vez no ciclo de vida do processador de registro para inicializá-lo e limpá-lo. Se a razão do desligamento for TERMINATE (porque o shard terminou devido a operações de divisão/merge), então o usuário também deve tomar cuidado para identificar todos os registros processados.

Implementamos a lógica de processamento de registro dentro do método process_records(). O código deve fazer um loop através do lote de registros e o checkpoint no final da chamada. O KCL assume que todos os registros foram processados. Em caso de falha na tarefa, a informação de checkpoint é utilizada pelo KCL para reiniciar o processamento do shard no último registro verificado.

O KCL conecta-se ao stream, enumera os shards e instancia um processador de registro para cada shard. Ele puxa registros de dados do fluxo e os encaminha para o processador de registro correspondente. O processador de registro também é responsável pelos registros processados no checkpoint.

Uma vez que cada processador de registro está associado a um shard único, vários processadores de registro podem ser executados em paralelo. Para tirar proveito das várias CPUs na máquina, cada processador de registro Python executa em um processo separado. Se executarmos a mesma aplicação KCL em várias máquinas, os processadores de registro passarão pelo balanceamento de carga juntamente com essas máquinas. Desta forma, o KCL permite alterar facilmente os tipos de máquinas ou o tamanho do fluxo.

Executando o exemplo

O release também vem com um exemplo de aplicação para contagem de palavras. Navegue até o diretório amazon_kclpy e instale o pacote.

$ python setup.py download_jars
$ python setup.py install

Um putter de exemplo é fornecido para criar uma stream no Kinesis chamado “words” e coloca palavras aleatórias nessa stream. Para iniciar o putter de exemplo, execute:

$ amazon_kclpy_helper.py --print_command --java 
--properties samples/sample.properties