谷歌雲數據流（Python）：讀取和寫入.csv文件的函數？

Question

我無法弄清楚GCP Dataflow Python SDK中讀寫csv文件（或任何非txt文件）的精確函數。對於BigQuery，我已經計算出以下函數：

beam.io.Read（beam.io.BigQuerySource（'％Table_ID％'）） beam.io.Write（beam.io.BigQuerySink（'％Table_ID ％'））

對於讀取文本文件，ReadFromText和WriteToText函數是我所知道的。

但是，我無法找到GCP Dataflow Python SDK的數據寫入或讀取csv文件的任何示例。請提供GCP Dataflow Python SDK函數來讀取和寫入csv文件，方法與我上面有關BigQuery的函數所做的相同。

Answer 1

CSV文件是文本文件。最簡單（雖然稍微不雅）的閱讀方法是做一個ReadFromText，然後拆分在逗號上閱讀的行（例如beam.Map(lambda x: x.split(','))）。

如需更優雅的選項，請查看this question，或者直接使用beam_utils pip存儲庫並使用beam_utils.sources.CsvFileSource源文件進行讀取。

Answer 2

beam_utils PiPy包中有一個CsvFileSource，它讀取.csv文件，處理文件頭並可以設置自定義分隔符。有關如何在this answer中使用此源的更多信息。希望有所幫助！