Ao escrever csv do CF para o bucket: 'with open(filepath, "w") as MY_CSV:' leva a "FileNotFoundError: [Errno 2] No such file or directory:"

Recebo esse erro FileNotFoundError: [Errno 2] No such file or directoryquando tento gravar um arquivo csv no bucket, usando um gravador csv que faz um loop em lotes de dados. O insight completo sobre os registros do Cloud Function em torno desse erro:

File "/workspace/main.py", line 299, in write_to_csv_file with
open(filepath, "w") as outcsv: FileNotFoundError: [Errno 2] No such
file or directory: 'gs://MY_BUCKET/MY_CSV.csv'

Function execution took 52655 ms, finished with status: 'crash' 

OpenBLAS WARNING - could not determine the L2 cache size on this
system, assuming 256k  ```

E isso, embora este bucket_filepath definitivamente exista: posso fazer upload de um arquivo fictício vazio e obter seu "gsutils URI" (clique com o botão direito do mouse nos três pontos no lado direito do arquivo) e o bucket_filepath terá a mesma aparência: 'gs://MY_BUCKET/MY_CSV.csv'.

Eu verifiquei salvar um dataframe de pandas fictício usando pd.to_csve funcionou com o mesmo bucket_filepath (!).

Portanto, deve haver outro motivo, provavelmente o gravador não é aceito, ou o with statementque abre o arquivo.

O código que gera o erro é o seguinte. É com o mesmo código trabalhando fora do Google Cloud Function em um cron job normal em um servidor local. Eu adicionei duas impressões de depuração ao redor da linha que lança o erro, o print("Right after opening the file ...")que não aparece mais. A subfunção query_execute_batch()que write_to_csv_file()está chamando para cada lote também é mostrada, mas provavelmente não é o problema aqui, pois o erro já ocorre no início ao abrir o arquivo csv.

requirements.txt(que são então importados como módulos):

SQLAlchemy>=1.4.2
google-cloud-storage>=1.16.1
mysqlclient==2.1.0
pandas==1.2.3
fsspec==2021.11.1
gcsfs==2021.11.1
unicodecsv==0.14.1

E a partir de main.py:

def query_execute_batch(connection):
    """Function for reading data from the query result into batches
    :yield: each result in a loop is a batch of the query result
    """
    results = execute_select_batch(connection, SQL_QUERY)
    print(f"len(results): {len(results)}")
    for result in results:
        yield result

def write_to_csv_file(connection, filepath):
    """Write the data in a loop over batches into a csv.
    This is done in batches since the query from the database is huge.
    :param connection: mysqldb connection to DB
    :param filepath: path to csv file to write data
    returns: metadata on rows and time
    """
    countrows = 0
    print("Right before opening the file ...")    
    with open(filepath, "w") as outcsv:
        print("Right after opening the file ...")        
        writer = csv.DictWriter(
            outcsv,
            fieldnames=FIELDNAMES,
            extrasaction="ignore",
            delimiter="|",
            lineterminator="\n",
        )
        # write header according to fieldnames
        writer.writeheader()

        for batch in query_execute_batch(connection):
            writer.writerows(batch)
            countrows += len(batch)
        datetime_now_save = datetime.now()
    return countrows, datetime_now_save

Lembre-se de que, para que o script acima funcione, eu importo, gcsfso que torna o bucket disponível para leitura e gravação. Caso contrário, eu provavelmente precisaria de um objeto de armazenamento em nuvem do Google, como por exemplo:

storage_client = storage.Client()
bucket = storage_client.bucket(BUCKET_NAME)

e, em seguida, faça o arquivo nesse bucket com outras funções, mas esse não é o objetivo aqui.

A seguir, o pd.to_csvcódigo que funciona, ele usa a saída de uma consulta SQL fictícia SELECT 1como entrada de um dataframe. Isso pode ser salvo no mesmo bucket_filepath, é claro que o motivo pode não ser apenas pd.to_csv()como tal, mas também que o conjunto de dados é um manequim em vez de strings unicode complexas de um enorme arquivo SELECT query. Ou há outra razão, estou apenas supondo.

if records is not None:
    df = pd.DataFrame(records.fetchall())
    df.columns = records.keys()
    df.to_csv(filepath,
        index=False,
    )
    datetime_now_save = datetime.now()
    countrows = df.shape[0]

Eu gostaria de usar o gravador csv para ter a chance de escrever em unicode com o módulo unicodecsv e a chance de usar os lotes.

Eu posso estar disposto a mudar para lotes ( loop + appendmode ou chunksize) em pandas como em Escrevendo dataframes grandes de Pandas para arquivo CSV em pedaços para se livrar desse problema de caminho de arquivo do bucket, mas eu gostaria de usar o código pronto (nunca toque em um sistema em execução ).

Como posso salvar esse csv com o gravador csv para que ele possa abrir um novo arquivo no bucket em writemode = with open(filepath, "w") as outcsv:?

A função fornecida write_to_csv_file()é apenas uma pequena parte da Cloud Function que usa uma ampla variedade de funções e funções em cascata. Não posso mostrar todo o caso reproduzível aqui e espero que possa ser respondido pela experiência ou por exemplos mais fáceis.