☰

При записи csv из CF в контейнер: 'with open(filepath, "w") as MY_CSV:' приводит к "FileNotFoundError: [Errno 2] Такого файла или каталога нет:"

tag-icon • tag-icon google-cloud-platform filesystems google-cloud-functions
При записи csv из CF в контейнер: 'with open(filepath, "w") as MY_CSV:' приводит к "FileNotFoundError: [Errno 2] Такого файла или каталога нет:"

Я получаю эту ошибку, FileNotFoundError: [Errno 2] No such file or directoryкогда пытаюсь записать файл csv в бакет, используя csv writer, который циклически обрабатывает пакеты данных. Полное представление о функциях облака регистрирует эту ошибку:


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  ```

И это, хотя этот bucket_filepath определенно существует: я могу загрузить пустой фиктивный файл и получить его «gsutils URI» (щелкните правой кнопкой мыши по трем точкам справа от файла), и bucket_filepath будет выглядеть так же: 'gs://MY_BUCKET/MY_CSV.csv'.

Я проверил сохранение фиктивного фрейма данных pandas вместо использования, pd.to_csvи это сработало с тем же bucket_filepath (!).

Следовательно, должна быть другая причина, вероятно, автор не принят, или тот, with statementкто открывает файл.

Код, который выдает ошибку, следующий. Это тот же код, работающий вне Google Cloud Function в обычном задании cron на локальном сервере. Я добавил два отладочных вывода вокруг строки, которая выдает ошибку, print("Right after opening the file ...")больше она не отображается. Подфункция query_execute_batch(), которая write_to_csv_file()вызывается для каждого пакета, также отображается, но, вероятно, проблема не в этом, поскольку ошибка возникает уже в самом начале при открытии-записи файла csv.

requirements.txt(которые затем импортируются как модули):

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

И из 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

Имейте в виду, что для работы скрипта выше я импортирую, gcsfsчто делает бакет доступным для чтения-записи. В противном случае мне, вероятно, понадобится объект облачного хранилища Google, например:

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

а затем создать файл в этом контейнере с дополнительными функциями, но это не является целью здесь.

В следующем pd.to_csvкоде, который работает, он использует вывод фиктивного SQL-запроса SELECT 1в качестве ввода для dataframe. Этоможетбыть сохранены в том же bucket_filepath, конечно, причина может быть не только pd.to_csv()в этом, но и в том, что набор данных является фиктивным, а не сложными строками unicode из огромного SELECT query. Или есть другая причина, я просто предполагаю.

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]

Я хотел бы использовать CSV-редактор, чтобы иметь возможность писать в Unicode с помощью модуля Unicodecsv и использовать пакеты.

Я мог бы перейти на пакеты ( loop + appendрежим или chunksize) в пандах, как вЗапись больших Pandas Dataframes в CSV-файл по частямчтобы избавиться от этой проблемы с путями к файлам ведра, но я бы предпочел использовать готовый код (никогда не трогайте работающую систему).

Как мне сохранить этот CSV-файл с помощью CSV-редактора, чтобы он мог открыть новый файл в контейнере в writeрежиме = with open(filepath, "w") as outcsv:?

Данная функция write_to_csv_file()— это всего лишь крошечная часть Cloud Function, которая использует широкий спектр функций и каскадных функций. Я не могу показать здесь весь воспроизводимый случай и надеюсь, что на него можно ответить с помощью опыта или более простых примеров.

решение1

Решение удивительное. Выдолженимпортируйте и используйте gcsfsмодуль, если вы хотите записать в файл с расширением open().

Если вы используете pd.to_csv(), import gcsfsне требуется, ноgcsfsвсе еще необходимо в том, requirements.txtчтобы сделать pd.to_csv()работу, поэтому панды, to_csv()похоже, используют его автоматически.

Оставив сюрприз pd.to_csv()в стороне, вот код, который отвечает на вопрос (проверено):

def write_to_csv_file(connection, filepath):
    """Write the QUERY result 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
    return: metadata on rows and time
    """
    countrows = 0
    print("Right before opening the file ...")
   

    # A gcsfs object is needed to open a file.
    # https://stackoverflow.com/questions/52805016/how-to-open-a-file-from-google-cloud-storage-into-a-cloud-function
    # https://gcsfs.readthedocs.io/en/latest/index.html#examples
    # Side-note (Exception):
    # pd.to_csv() needs neither the gcsfs object, nor its import.
    # It is not used here, but it has been tested with examples.
    fs = gcsfs.GCSFileSystem(project=MY_PROJECT)
    fs.ls(BUCKET_NAME)


    # wb needed, else "builtins.TypeError: must be str, not bytes"
    # https://stackoverflow.com/questions/5512811/builtins-typeerror-must-be-str-not-bytes
    with fs.open(filepath, 'wb') as outcsv:
        print("Right after opening the file ...")

        writer = csv.DictWriter(
            outcsv,
            fieldnames=FIELDNAMES,
            extrasaction="ignore",
            delimiter="|",
            lineterminator="\n",
        )
        # write header according to fieldnames
        print("before writer.writeheader()")
        writer.writeheader()
        print("after 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

Примечание

Не используйте CSV-редактор подобным образом.

Это занимает слишком много времени, вместо pd.to_csv()с chunksizeпараметром 5000, которому требуется всего 62 секунды для загрузки 700 тыс. строк и сохранения их в виде csv в бакете, CF с записывающим пакеты занимает более 9 минут, что превышает лимит времени ожидания. Поэтому я вынужден использовать pd.to_csv()вместо этого и преобразовывать свои данные в dataframe для этого.

Связанный контент