☰

CF からバケットに csv を書き込む場合: 'with open(filepath, "w") as MY_CSV:' は "FileNotFoundError: [Errno 2] No such file or directory:" になります

tag-icon • tag-icon google-cloud-platform filesystems google-cloud-functions
CF からバケットに csv を書き込む場合: 'with open(filepath, "w") as MY_CSV:' は "FileNotFoundError: [Errno 2] No such file or directory:" になります

FileNotFoundError: [Errno 2] No such file or directoryデータのバッチをループする csv ライターを使用して、バケットに csv ファイルを書き込もうとすると、このエラーが発生します。このエラーに関する Cloud Function ログの完全な情報は次のとおりです。


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」を取得すると (ファイルの右側にある 3 つのドットを右クリック)、bucket_filepath は同じになります'gs://MY_BUCKET/MY_CSV.csv'。

代わりにダミーの pandas データフレームを保存して確認したところpd.to_csv、同じ bucket_filepath (!) で動作しました。

したがって、別の理由があるはずです。おそらく、ライターが受け入れられていないか、with statementファイルを開く が原因と考えられます。

エラーをスローするコードは次のとおりです。これは、ローカル サーバーの通常の cron ジョブで Google Cloud Function の外部で動作する同じコードです。エラーをスローする行の周囲に 2 つのデバッグ プリントを追加しましたが、エラーは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 Cloud Storage オブジェクトが必要になる可能性があります。

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

そして、そのバケット内にさらなる機能を備えたファイルを作成しますが、それはここでの目的ではありません。

以下のpd.to_csvコードは、ダミーSQLクエリの出力をSELECT 1データフレームの入力として使用します。できる同じ 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 ライターを使用して、unicodecsv モジュールで Unicode で書き込む機会とバッチを使用する機会を得たいと思います。

私はパンダのバッチ（loop + appendモードまたは）に変更してもよいと思う。chunksize大きな Pandas データフレームをチャンクで CSV ファイルに書き込むこのバケット ファイルパスの問題を解決するには、既製のコード (実行中のシステムには絶対に触れない) を使用することをお勧めします。

writecsv ライターを使用して csv を保存し、モード =でバケット内の新しいファイルを開くにはどうすればよいですかwith open(filepath, "w") as outcsv:?

指定された関数は、write_to_csv_file()幅広い関数とカスケードされた関数を使用する Cloud Functions のほんの一部です。ここでは再現可能なケース全体を示すことはできませんが、経験やより簡単な例によって回答できることを願っています。

答え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 ライターを使用しないでください。

時間がかかりすぎます。パラメータ 5000 ではpd.to_csv()、chunksize70 万行をロードしてバケットに CSV として保存するのに 62 秒しかかかりませんが、バッチ ライターを使用した CF では 9 分以上かかり、タイムアウト制限を超えています。そのため、pd.to_csv()代わりに を使用し、データをデータフレームに変換する必要があります。

関連情報