输入文件未从 pd.read_csv 读取

Question

我正在尝试使用 pandas 从 apache Beam 读取存储在 google 存储中的文件，但出现错误

def Panda_a(self):
    import pandas as pd
    data = 'gs://tegclorox/Input/merge1.csv'
    df1 = pd.read_csv(data, names = ['first_name', 'last_name', 'age', 
         'preTestScore', 'postTestScore'])
    return df1
ip2 = p |'Split WeeklyDueto' >> beam.Map(Panda_a)
ip7 = ip2 | 'print' >> beam.io.WriteToText('gs://tegclorox/Output/merge1234')

当我执行上述代码时，错误提示路径不存在。知道为什么吗？

Answer 1

这段代码有很多问题。

• 试图让 Pandas 从 Google Cloud Storage 读取文件。 Pandas 不支持 Google Cloud Storage 文件系统（正如 @Andrew 指出的那样 - documentation 表示支持的方案是 http、ftp、s3、file）。但是，您可以使用 Beam FileSystems.open() API 获取文件对象，并将该对象而不是文件路径提供给 Pandas。
• p | ... >> beam.Map(...) - beam.Map(f) 使用给定函数 f 转换输入 PCollection 的每个元素，它不能应用于管道本身。在您的情况下，您似乎只想在没有任何输入的情况下运行 Pandas 代码。您可以通过提供虚假输入来模拟它，例如beam.Create(['ignored'])
• beam.Map(f) 要求 f 返回单个值（或者更像：如果它返回一个列表，它将将该列表解释为单个值），但是您的代码为其提供了一个返回 Pandas 数据帧的函数。我强烈怀疑您是否要创建一个包含单个元素的PCollection，其中该元素是整个数据框 - 更有可能的是，您希望数据框的每一行都有一个元素。为此，您需要使用beam.FlatMap，并且需要df.iterrows() 或类似的名称。

一般来说，我完全不知道为什么要使用 Pandas 读取 CSV 文件。您可以使用 Beam 的 ReadFromText 和 skip_header_lines=1 阅读它，然后自己解析每一行 - 如果您有大量数据，这将更有效率（如果您只有少量数据并执行不要期望它变得大到足以超过单台机器的能力 - 比如说，如果它永远不会超过几 GB - 那么 Beam 是错误的工具）。