我有一个用例，需要T从DoFn 输出多个。所以DoFn函数返回一个PCollection<List<T>>。我想将其转换为PCollection<T>以便稍后在管道中进行过滤，例如：
PCollection<T> filteredT = filterationResult.apply(Filter.byPredicate(p -> p.equals(T) == T));
目前，我能想到的最好的方法是，List<T>从ParDo我KV<String,List<T>>用相同键返回的函数返回每个项目。然后在管道中，我可以执行以下操作以合并结果：
filterationResult.apply("Group", GroupByKey.<String, List<T>>create())
或者可以拨打我c.output(T)从DOFN（这里c是ProcessContext对象传入）多次？
我编写了一个Dataflow作业，当我手动运行它时效果很好。这是相关的部分（为清楚起见，删除了一些验证代码）：
parser.add_argument('--end_datetime',
dest='end_datetime')
known_args, pipeline_args = parser.parse_known_args(argv)
query = <redacted SQL String with a placeholder for a date>
query = query.replace('#ENDDATETIME#', known_args.end_datetime)
with beam.Pipeline(options=pipeline_options) as p:
rows = p | 'read query' >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True))
现在，我想创建一个模板并将其安排为以动态ENDDATETIME定期运行。据我了解，为了做到这一点，我需要根据此文档将add_argument更改为add_value_provider_argument：
https://cloud.google.com/dataflow/docs/templates/creating-templates
不幸的是，似乎ValueProvider值在我需要它们时不可用，它们仅在管道内部可用。（如果我在这里错了，请纠正我...）。所以我有点卡住了。
是否有人对如何在Dataflow模板中的查询中获取动态日期有任何建议？
python 依赖项可以加载到 Google Cloud Dataflow 管道中吗？
我想使用gensim 的短语建模器，它逐行读取数据以自动检测常用短语/二元组（经常出现在彼此旁边的两个词）。
因此，管道的第一次运行会将每个句子传递给这个短语建模器。
然后第二次通过管道将采用相同的短语建模器并将这个短语建模器应用于每个句子，以确定应该一起建模的短语。例子：
machine和learning频繁地出现在语料库中，它们将被转换为单个单词machine_learning。 这可以在 Dataflow 中完成吗？
可以pip install gensim在工作机器上强制传递构建/需求文件吗？
我正在使用光束管道来处理 json 并将其写入 bigquery。JSON 是这样的。
{
"message": [{
"name": "abc",
"itemId": "2123",
"itemName": "test"
}, {
"name": "vfg",
"itemId": "56457",
"itemName": "Chicken"
}],
"publishDate": "2017-10-26T04:54:16.207Z"
}
我使用 Jackson 将其解析为以下结构。
class Feed{
List<Message> messages;
TimeStamp publishDate;
}
public class Message implements Serializable{
/**
*
*/
private static final long serialVersionUID = 1L;
private String key;
private String value;
private Map<String, String> eventItemMap = new HashMap<>();
this property translate the list of map as a single map with all the key-value pair …google-bigquery google-cloud-dataflow apache-beam apache-beam-io
我有一个将结果写入 BigQuery 表的 Apache Beam/Dataflow 管道。然后我想查询此表以获取管道的单独部分。但是，我似乎无法弄清楚如何正确设置此管道依赖项。我编写（然后想要查询）的新表与用于某些过滤逻辑的单独表相连，这就是为什么我实际上需要编写表然后运行查询。逻辑如下：
with beam.Pipeline(options=pipeline_options) as p:
table_data = p | 'CreatTable' >> # ... logic to generate table ...
# Write Table to BQ
table_written = table_data | 'WriteTempTrainDataBQ' >> beam.io.WriteToBigQuery(...)
query_results = table_written | 'QueryNewTable' >> beam.io.Read(beam.io.BigQuerySource(query=query_new_table))
ifquery_new_table实际上是对已经存在的 BQ 表的查询，我更改为query_results = p |而不是table_written这正常工作。但是，如果我尝试查询我在管道中间写入的表，那么在实际生成该表之前，我无法让管道步骤“等待”。有没有办法做到这一点，我忽略了？
当我尝试按顺序执行此步骤时，我收到一个断言错误assert isinstance(pbegin, pvalue.PBegin) AssertionError，我正在阅读这意味着这table_written是问题，因为它不是有效的 PCollection 实例。
有谁知道我可以用什么来代替 table_written 来让它按需要按顺序运行？
我试图使用Apache Beam创建一个Dataflow管道,我无法按照文档,找不到任何示例.
管道很简单.
目前,我陷入了第2步.我找不到任何关于如何从pub/sub读取并使用它的示例.
这是我到目前为止所希望的代码
class ExtractFlowInfoFn extends DoFn<PubsubMessage, KV<String, String>> {
public void processElement(ProcessContext c) {
KV.of("key", "value");
}
}
public static void main(String[] args) {
Pipeline p = Pipeline.create(
PipelineOptionsFactory.fromArgs(args).withValidation().create());
p.apply("ReadFromPubSub", PubsubIO.readMessages().fromSubscription("test"))
.apply("ConvertToKeyValuePair", ParDo.of(new ExtractFlowInfoFn()))
.apply("WriteToLog", ));
};
我通过以下几个例子来提出代码.说实话,我不知道我在这做什么.
请帮助我理解这一点或将我链接到正确的文档.
我有一个可以在本地执行而没有任何错误的管道。我曾经在本地运行的管道中遇到此错误
'Clients have non-trivial state that is local and unpickleable.'
PicklingError: Pickling client objects is explicitly not supported.
我相信我通过降级到 apache-beam=2.3.0 解决了这个问题，然后在本地它会完美运行。
现在我正在使用DataflowRunner 并且在 requirements.txt 文件中我有以下依赖项
apache-beam==2.3.0
google-cloud-bigquery==1.1.0
google-cloud-core==0.28.1
google-cloud-datastore==1.6.0
google-cloud-storage==1.10.0
protobuf==3.5.2.post1
pytz==2013.7
但我又犯了这个可怕的错误
'Clients have non-trivial state that is local and unpickleable.'
PicklingError: Pickling client objects is explicitly not supported.
为什么它给我 DataflowRunner 的错误而不是 DirectRunner？他们不应该使用相同的依赖项/环境吗？任何帮助，将不胜感激。
我读过这是解决它的方法，但是当我尝试它时，我仍然遇到相同的错误
class MyDoFn(beam.DoFn):
def start_bundle(self, process_context):
self._dsclient = datastore.Client()
def process(self, context, *args, **kwargs):
# do stuff with self._dsclient
来自https://github.com/GoogleCloudPlatform/google-cloud-python/issues/3191
我之前在本地修复此问题的参考帖子：
根据Beam网站,
通常,对管道代码执行本地单元测试比调试管道的远程执行更快更简单.
我想为我的Beam/Dataflow应用程序使用测试驱动开发,因为这个原因写入Bigtable.
然而,遵循Beam测试文档我陷入了僵局--PAssert没用,因为输出PCollection包含org.apache.hadoop.hbase.client.Put对象,它们不会覆盖equals方法.
我无法获得 PCollection 的内容来对它们进行验证,因为
不可能直接获取PCollection的内容 - Apache Beam或Dataflow管道更像是应该进行什么处理的查询计划,其中PCollection是计划中的逻辑中间节点,而不是包含数据.
那么除了手动运行之外,我该如何测试这个管道呢？我正在使用Maven和JUnit(在Java中,因为所有Dataflow Bigtable Connector似乎都支持).
使用 ndjson 格式的文本文件，以下代码产生了我所期望的。一个 ndjson 文件，其中quotes.USD dict 未嵌套并删除了原始quotes 元素。
def unnest_quotes(element):
element['USDquotes'] = element['quotes']['USD']
del element['quotes']
return element
p = beam.Pipeline(options=pipeline_options)
ReadJson = p | ReadFromText(known_args.input,coder=JsonCoder())
MapFormattedJson = ReadJson | 'Map Function' >> beam.Map(unnest_quotes)
MapFormattedJson | 'Write Map Output' >> WriteToText(known_args.output,coder=JsonCoder())
但是，当我尝试使用 ParDo 实现相同的目标时，我不理解这种行为。
class UnnestQuotes(beam.DoFn):
def process(self,element):
element['USDquotes'] = element['quotes']['USD']
del element['quotes']
return element
p = beam.Pipeline(options=pipeline_options)
ReadJson = p | ReadFromText(known_args.input,coder=JsonCoder())
ClassFormattedJson = ReadJson | 'Pardo' >> beam.ParDo(UnnestQuotes())
ClassFormattedJson | 'Write Class Output' >> WriteToText(known_args.output,coder=JsonCoder())
这将生成一个文件，其中 dict 的每个键位于单独的行上，没有值，如下所示。
"last_updated" …