Erro ao rodar descompata casos de dengue | Apache Beam: Data Pipeline com Python

Ola. Quando rodo o programa, aparece um erro na funcao "casos_dengue". Tentei achar onde esta errado mas nao consegui. Meu programa e':

import apache_beam as beam from apache_beam.io import ReadFromText from apache_beam.options.pipeline_options import PipelineOptions

pipeline_options = PipelineOptions(argv=None) pipeline = beam.Pipeline(options=pipeline_options) 0 colunas_dengue = [ 'id', 'data_iniSE', 'casos', 'ibge_code', 'cidade', 'uf', 'cep', 'latitude', 'longitude']

def lista_para_dicionario(elemento,colunas): """ Recebe duas listas e Retorna um dicionario """ return dict(zip(colunas,elemento))

def texto_para_lista(elemento, delimitador='|'): """ Recebe um texto e um delimitador e retorna uma lista de elementos pelo delimitador """ return elemento.split(delimitador)

def trata_datas(elemento): """ Recebe um dicionario e cria um novo campo com ANO-MES Retorna o mesmo dicionario com o novo campo """ elemento['ano_mes'] = '-'.join(elemento['data_iniSE'].split('-')[:2]) return elemento

def chave_uf(elemento): """ Recebe um dicionario e retorna uma tupla com o estado e o elemento (UF, dicionario) """ chave = elemento['uf'] return (chave, elemento)

def casos_dengue(elemento): """ Rece uma tupla (RS, [{},{}]) Retorna uma tupla (RS-2024-12, 8.0) """ uf, registros = elemento for registro in registros: yield (f"{uf}-{registro['ano_mes']}",registro['casos'])

dengue = ( pipeline | "Leitura do dataset de dengue" >> ReadFromText('casos_dengue.txt', skip_header_lines = 1) | "De texto para lista" >> beam.Map(texto_para_lista) | "De lista para dicionario" >> beam.Map(lista_para_dicionario,colunas_dengue) | "Criar campo ano_mes" >> beam.Map(trata_datas) | "Criar chave pelo estado" >> beam.Map(chave_uf) | "Agrupar pelo estado" >> beam.GroupBy() | "Descompactar casos de dengue" >> beam.FlatMap(casos_dengue) | "Mostrar resultados" >> beam.Map(print) )

pipeline.run()

E o erro apresentado e':

WARNING:apache_beam.options.pipeline_options:Discarding invalid overrides: {'argv': None} Traceback (most recent call last): File "apache_beam/runners/common.py", line 1418, in apache_beam.runners.common.DoFnRunner.process File "apache_beam/runners/common.py", line 624, in apache_beam.runners.common.SimpleInvoker.invoke_process File "apache_beam/runners/common.py", line 1572, in apache_beam.runners.common.OutputHandler.handleprocess_outputs File "main.py", line 53, in casos_dengue yield (f"{uf}-{registro['ano_mes']}",registro['casos']) TypeError: tuple indices must be integers or slices, not str

During handling of the above exception, another exception occurred:

Traceback (most recent call last): File "main.py", line 67, in pipeline.run() ... File "main.py", line 53, in casos_dengue yield (f"{uf}-{registro['ano_mes']}",registro['casos']) TypeError: tuple indices must be integers or slices, not str [while running 'Descompactar casos de dengue']

Olá, Jessica! Tudo bem?

Aparentemente, o erro está na hora de passar o valor. Pelo comentário que você deixou na função, você indica que o valor de elemento contém o formato (RS, [{},{}]). O problema está no segundo campo dessa tupla.

Nesse caso, o segundo valor o campo (RS, [{},{}]) contém uma lista com duas chaves vazias separadas por vírgula.

No caso da sua função, você deve utilizar apenas um dicionário dentro da lista contendo as duas chaves, de ano_mes e casos. Dessa forma, o código ficaria assim:

def casos_dengue(elemento): 
    """ Receba uma tupla (RS, [{}]) e retorne uma tupla (RS-2024-12, 8.0) """ 
    uf, registros = elemento 
    for registro in registros: 
        print(f"{uf}-{registro['ano_mes']}", registro['casos'])

elemento = [{'ano_mes': '2024-12', 'casos': 8}]

casos_dengue(('RS', elemento))

Espero ter ajudado, Jéssica. Se tiver mais alguma dúvida, estou à disposição. Bons estudos =)

Caso este post tenha lhe ajudado, por favor, marcar como solucionado ✓. Bons Estudos!