1

Estou tentando criar uma classe (Processor) que executa um serie de funcoes para obter resultados de uma API Estado a Estado. (STATE)

1- READ dados de uma tabela.
2- SEND request para a API para processar.
3- WAIT jobs to be complete.
4- DOWNLOAD dos resultados quando completos.

Gostaria de implementar concurrencia de modo a que pode-se correr para varios estados ao mesmo tempo . Estava tentanto algo parecido com o que Luciano Ramalho implementa com a lib concurrent.futures em Fluent Python .( Chapter 17. Concurrency with Futures) mas nao consigo a concorrencia

ERROR:
    res = executor.map(run_single(), sorted(states))
    TypeError: run_single() missing 1 required positional argument: 'state'

Mas quando incluo state em run single apenas corre sequencialmente estado a estado.

Estou usando python 3.5.~ Parte do meu codigo em baixo - Agradecido por toda e qualquer orientaçao.

        import datetime
        import os
        import time
        import xml.etree.ElementTree as ET
        import psycopg2
        import psycopg2.extras
        import requests
        from concurrent import futures


        class Processor(object):    
            database = 'db'
            user = 'user'
            password = 'password'

            def __init__(self, state):
                self.base_url = 'api.com'
                self.state = state
                self.status = None
                self.fetch_size = 1000000
                self.job_id = ''

                self.send_requests(self.state)

            def send_requests(self, state):
                payload = dict(params)

                # connection to postgres db table , fetch data.
                conn = psycopg2.connect(
                    "dbname='%s' user='%s' host='host' password='%s'" % (database, user, password))
                cursor = conn.cursor('%s' % state, cursor_factory=psycopg2.extras.DictCursor)
                sql = ("select * from table where state='%s' limit 1" % state)
                cursor.execute(sql)

                try:
                    # function to build/send requests fetching data by chunks of fetch_size limited.
                    while True:
                        fetchs = cursor.fetchmany(self.fetch_size)
                        if len(fetchs) != 0:
                            chunk = ''
                            for fetch in fetchs:
                                try:
                                    row = fetch[0] + '|' + fetch[1] + '|' + fetch[2] + '\n'
                                    chunk += row
                                except:
                                    print('>ERROR ->', fetch[0])
                                    pass
                            header = 'header\n'
                            row = requests.post(self.base_url, params=payload, data=header + chunk)
                            response = row.text
                            print('-> %s: response job_xml: %s' % (state, response))
                            root = ET.fromstring(response)
                            self.job_id = root.find('Response/MetaInfo/RequestId').text
                            print('-> %s: response job_id: %s' % (state, self.job_id))
                            self.check_jobs(state)
                        else:
                            break
                except Exception as e:
                    print(e)
                    pass

                # Function checking the status of the job_id if completed download() the results if not wait and retry.
            def check_jobs(self, state):
                print('->>> %s: Checking job %s  <<<-' % (state, self.job_id))
                status = self.get_status(self.job_id)
                if status == 'completed':
                    print('-> %s: status: %s, job_id: %s  ' % (state, status, self.job_id))
                    self.download_results(self.job_id)
                else:
                    time.sleep(4)  # 480 large million requests
                    self.check_jobs(state)

                # Function to return status of job_id
            def get_status(self, job_id):
                url_status = 'url that get status of job_id'
                req_status = requests.get(url_status)
                root = ET.fromstring(req_status.text)
                status = root.find('Response/Status').text
                return status

                # Function download the results
            def download_results(self, job_id):
                url_download = 'url to download job_id'
                print('-> %s: downloading jod_id: %s @ URL [%s]' % (self.state, job_id, url_download))
                r = requests.get(url_download, stream=True)

                # create folder for state if not exists
                download = os.path.join(self.responses_folder, self.state)
                if not os.path.exists(download):
                    os.makedirs(download)

                # Save result to folder
                save_as = os.path.join(download, str(job_id + '.zip'))
                with open(save_as, 'wb') as f:
                    for chunk in r.iter_content(chunk_size=1024):
                        if chunk:
                            f.write(chunk)

                print('-> %s: downloaded job_id: %s @ folder [ %s ] ' % (self.state, job_id, save_as))
                self.delete_results(job_id)


if __name__ == "__main__":
    states = ['AK', 'AL', 'AR']
    workers = 20

    def run_single(state):
        Processor(state)

    for state in states:
        with futures.ThreadPoolExecutor(workers) as executor:
            res = executor.map(run_single(), sorted(states))

1 Resposta 1

1

O seu problema é que ao criar uma tarefa no seu Executor você está chamando a funçao, quando usa a expressão run_single(). Nesse caso a pergunta não está muito bem colocada, ou que tem fragmentos do seu código que não permitem ver o problema em si - são partesso, o programa chamaria a função e o valor de retorno da mesma é que seria passado para a chamada ao Executor.map - e daí vem o seu erro: a função requer um parâmetro posicional (state), que não é passado.

O que você deve fazer é passar a própria função como parâmetro para o Executor - que então se encarrega de chama-la dentro de uma thread separada. Para isso, você não deve colocar parênteses depois do nome da função (o que em Python faz com que ela seja tratada como qualquer outro objeto e simplesmente passada como parâmetro).

Em resumo, basta reescrever sua linha de chamada do executor como:

res = executor.map(run_single, states)

Perceba que eu removi o sorted também - o executor não tem como garantir a ordem em que cada uma das tarefas será processada, então tentar ordenar a lista de entrada só pode te passar a falsa impressão de que as tarefas seriam executadas em alguma ordem espécífica. É melhor deixar claro que essa ordem é arbitrária.

(O nome da classe também está errado dentro de run_single - deve ser Processor, não Geocode)

E por fim, voce está criando uma nova conexação ao banco para cada tarefa que está executando - em geral isso não é uma boa prática - criar uma conexão ao banco é uma tarefa relativamente custosa - o melhor é usar algum tipo de pool de conexões, ou usar apenas uma conexação global e criar um cursor a partir da mesma dentro da sua classe.

3
  • obrigado por todas a dicas - vou dar um look no coding e tentar implementar sua sugestao , agradecido tb pela dica das connection pools vou ler sobre o assunto . Volto já. Commented 5/10/2016 às 9:49
  • jsbueno - funcionou no entanto esperava ver varios python.exe a serem executados ao mesmo tempo, verifiquei apenas 1. E no final o tempo de execução nao foi grande surpresa - foi igual ou inferior ao sequencial - alguma sugestao do que poderá estar a acontecer ? Commented 12/10/2016 às 11:01
  • Voce está usando o ThreadPoolEecutor - vai ter um processo só mesmo. Para rodar em multiplos processos, use o ProcessPoolExecutor (também em concurrent.futures). Claro que não tenho como só olhar o seu código e identificar os gargalos - mas aparentemente é código com abstante IO que deveria sim ter se beneficiado com a execução em threads paralelas - exceto pelo uso da biblioteca ElementTree para processamento de XML: ela é em Python puro, e sua execução acaba não sendo paralelizável em threads. Se esse for o problema, usar o ProcessPoolExecutor pode ser o suficiente.
    – jsbueno
    Commented 12/10/2016 às 13:51

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .