Google Cloud Dataflow 访问云存储上的 .txt 文件答案

【问题标题】：Google Cloud Dataflow access .txt file on cloud storageGoogle Cloud Dataflow 访问云存储上的 .txt 文件
【发布时间】：2017-10-10 22:24:57
【问题描述】：

如果我在 GCS 上存储了一个 .txt 文件，其中包含将用作 beam.Filter 一部分的单词列表，是否可以在我的 apache 光束管道中动态访问此列表？我知道我可以将此列表定义为管道中的全局变量，但我不确定如何将整个文件读入列表以及是否有任何光束技巧来完成此操作。有什么建议么？这是我当前的实现，它不起作用..

def boolean_terms(word, term_list):
  if word in term_list:
    return (word, 1)
  else:
    return (word, 0)

# side table
filter_terms = p | beam.io.ReadFromText(path_to_gcs_txt_file)

words = ...

filtered_words = words | beam.FlatMap(lambda x: 
    [boolean_terms(word, filter_terms) for word in x])

我收到以下错误“TypeError：'_InvalidUnpickledPCollection' 类型的参数不可迭代”

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

您可以通过side input 访问单词列表。我相信beam.Filter 转换支持使用来自过滤器函数的侧输入，其方式与该链接示例中的FlatMap 和ParDo 完全相同。

类似：

words | beam.Filter(lambda x, filter_terms: word in filter_terms,
                    filter_terms=pvalue.AsList(p | beam.io.ReadFromText(path)))

【讨论】：

谢谢！我想我更接近了，但它似乎仍然不适合我。我错过了什么吗？
啊，我想我明白了——我需要添加 pvalue.AsList(filter_terms) 才能正常工作