我正在开发工作流流程，以使用Apache Beam的Python SDK在Google Cloud Dataflow上运行。
在本地运行时，工作流成功完成，没有任何错误，并且数据输出完全符合预期。
当我尝试在Dataflow服务上运行时，它引发以下错误：
AttributeError: '_UnwindowedValues' object has no attribute 'sort'
来自以下代码：
class OrderByDate(beam.DoFn):
def process(self, context):
(k, v) = context.element
v.sort(key=operator.itemgetter('date'))
return [(k, v)]
并使用如下标准调用该方法beam.ParDo：
'order_by_dates' >> beam.ParDo(OrderByDate())
(k, v)元组中的数据如下例所示：
('SOME CODE', {'date':'2017-01-01', 'value':1, 'date':'2016-12-14', 'value':4})
随着v是日期和值的对象
我尝试过切换到标准lambda函数也会引发相同的错误。
有什么想法为什么它在本地运行与在Dataflow上运行不同？或建议解决。
有没有之间的差异ParDo,并FlatMap在数据流/ Apache的梁？
我认为两者都将一个函数应用于传入的每个元素PCollection,并返回迭代; 但我想必须有一些区别？
有没有办法将侧输入应用于 Apache Beam 中的 BigQueryIO.read() 操作。
举例来说，我在 PCollection 中有一个值，我想在查询中使用它来从 BigQuery 表中获取数据。这可以使用侧面输入吗？或者在这种情况下应该使用其他东西吗？
我在类似的情况下使用了 NestedValueProvider，但我想我们只能在某个值取决于我的运行时值时使用它。或者我可以在这里使用相同的东西吗？如果我错了，请纠正我。
我试过的代码：
Bigquery bigQueryClient = start_pipeline.newBigQueryClient(options.as(BigQueryOptions.class)).build();
Tabledata tableRequest = bigQueryClient.tabledata();
PCollection<TableRow> existingData = readData.apply("Read existing data",ParDo.of(new DoFn<String,TableRow>(){
@ProcessElement
public void processElement(ProcessContext c) throws IOException
{
List<TableRow> list = c.sideInput(bqDataView);
String tableName = list.get(0).get("table").toString();
TableDataList table = tableRequest.list("projectID","DatasetID",tableName).execute();
for(TableRow row:table.getRows())
{
c.output(row);
}
}
}).withSideInputs(bqDataView));
我得到的错误是：
Exception in thread "main" java.lang.IllegalArgumentException: unable to serialize BeamTest.StarterPipeline$1@86b455
at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:53)
at org.apache.beam.sdk.util.SerializableUtils.clone(SerializableUtils.java:90)
at org.apache.beam.sdk.transforms.ParDo$SingleOutput.<init>(ParDo.java:569)
at org.apache.beam.sdk.transforms.ParDo.of(ParDo.java:434)
at BeamTest.StarterPipeline.main(StarterPipeline.java:158)
Caused by: java.io.NotSerializableException: com.google.api.services.bigquery.Bigquery$Tabledata …我看过很多关于同一主题的问题。但是，我仍然遇到写入 GCS 的问题。我正在从 pubsub 阅读该主题并尝试将其推送到 GCS。我已经提到了这个链接。但是，在最新的束包中找不到 IOChannelUtils。
PCollection<String> details = pipeline
.apply(PubsubIO.readStrings().fromTopic("/topics/<project>/sampleTopic"));
PCollection<KV<String, String>> keyedStream = details.apply(WithKeys.of(new SerializableFunction<String, String>() {
public String apply(String s) {
return "constant";
}
}));
PCollection<KV<String, Iterable<String>>> keyedWindows = keyedStream.apply(Window.<KV<String, String>>into(FixedWindows.of(ONE_MIN)).withAllowedLateness(ONE_DAY)
.triggering(AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(10))
.withLateFirings(AfterFirst.of(AfterPane.elementCountAtLeast(10),
AfterProcessingTime.pastFirstElementInPane().plusDelayOf(TEN_SECONDS))))
.discardingFiredPanes()).apply(GroupByKey.create());
PCollection<Iterable<String>> windows = keyedWindows.apply(Values.create());
这是我从堆栈溢出中的许多其他类似主题中获取的。现在，我明白了，TextIO 确实支持 withWindowedWrites 和 withNumShards 的无界 PCollection 写入选项。
参考：使用 DoFn 使用 Cloud Dataflow 从 PubSub 写入 Google Cloud Storage
但是，我不明白，我该怎么做。
我正在尝试按如下方式写入 GCS。
FilenamePolicy policy = DefaultFilenamePolicy.constructUsingStandardParameters(
StaticValueProvider.of(outputDirectory), DefaultFilenamePolicy.DEFAULT_SHARD_TEMPLATE, "");
details.apply(TextIO.write().to("gs://<bucket>/topicfile").withWindowedWrites()
.withFilenamePolicy(policy).withNumShards(4));
我没有足够的点数来为 Stack Overflow …
在分布式处理环境中,通常使用"部分"文件名,例如"part-000",是否可以编写某种扩展名来重命名各个输出文件名(例如每个窗口文件名) Apache Beam？
为此,可能必须能够为窗口指定名称或根据窗口的内容推断文件名.我想知道这种方法是否可行.
至于解决方案应该是流式还是批量式,流式模式示例是优选的
我目前正在尝试根据数据中包含的特定密钥将在Google Dataflow上运行的Beam管道分成多个目标.当使用TaggedOutput标记对"fork"的每个端点进行硬编码时,我能够正常工作.但是,在将来,我不会总是知道底层数据中存在哪些键,因此我希望使用类似于下面的for循环动态创建流程中的后续步骤:
p = beam.Pipeline(options=pipeline_options)
pipe = p | 'ReadFromGCS' >> ReadFromText(args['input']) \
| 'TagAllLines' >> beam.ParDo(produce_tagged_output_keys).with_outputs()
for client in pipe:
client = pipe.client | client+'MapLinesToDicts' >> beam.Map(lambda line: dict(record=line)) \
| client+'WriteToBQTable' >> WriteToBigQuery(client+'_test', dataset=bq_dataset, project=project_id, schema='record:string')
我的理解是结果 .with_outputs()应该是可迭代的,不是吗？当我运行上面的命令时,它会pipe毫无问题地执行,但完全忽略for循环.有没有办法动态地做到这一点,我错过了？
我们正在使用 Python SDK 在 google 数据流中运行日志文件解析作业。数据分布在数百条每日日志中，我们通过 Cloud Storage 中的文件模式读取这些日志。所有文件的数据量约为 5-8 GB（gz 文件），总共有 50-80 百万行。
loglines = p | ReadFromText('gs://logfile-location/logs*-20180101')
此外，我们有一个简单的（小）映射 csv，它将日志文件条目映射到人类可读的文本。大约有 400 行，5 kb 大小。
例如，带有 [param=testing2] 的日志文件条目应映射到最终输出中的“客户请求 14 天免费产品试用”。
我们在一个带有 sideinput 的简单 beam.Map 中执行此操作，如下所示：
customerActions = loglines | beam.Map(map_logentries,mappingTable)
其中 map_logentries 是映射函数，mappingTable 是映射表。
然而，这只有在我们通过 open() / read() 读取原生 python 中的映射表时才有效。如果我们通过 ReadFromText() 使用光束管道执行相同的操作，并将生成的 PCollection 作为侧输入传递给 Map，如下所示：
mappingTable = p | ReadFromText('gs://side-inputs/category-mapping.csv')
customerActions = loglines | beam.Map(map_logentries,beam.pvalue.AsIter(mappingTable))
性能完全分解为每秒大约 2-3 个项目。
现在，我的问题：
对我们来说，映射确实经常变化，我需要找到一种方法让“普通”用户提供它。这个想法是在 Cloud Storage 中提供映射 …
我的管道的这部分应该接受一个输入，对其应用适当的元组标记，然后根据它接收到的标记对输入进行进一步处理。
运行以下代码时，主标签 (tag1) 中的 PCollection 工作正常。但是，附加标签（tag2、tag3）将在以下位置抛出此错误.apply()：
线程“main”中的异常 java.lang.IllegalStateException：无法为 Assign Output.out1 [PCollection] 返回默认编码器。更正以下根本原因之一： 没有手动指定编码器；您可以使用 .setCoder() 这样做。从 CoderRegistry 推断编码器失败：无法为 V 提供编码器。
为什么这个错误发生在 tag2 而不是 tag1？请注意，如果我将 tag2 设为主要输出，将 tag1/tag3 作为附加输出并适当地重新排序代码，则 tag2 处理成功，但 tag1/tag3 将引发错误。
主管道：
PCollectionTuple pct = outputPair.apply("Assign Output", ParDo.of( new output())
.withOutputTags(output.tag1, TupleTagList.of(output.tag2).and(output.tag3)));
//Tag1 Output
PCollection<KV<String, outResultPair>> tagPair1 = pct.get(output.tag1)
.apply("Process", ParDo.of( new ABCOutput()))
//Tag2 Output
PCollection<KV<String, outResultPair>> tagPair2 = pct.get(output.tag2)
.apply("Process", ParDo.of( new DEFOutput())) //Error Thrown here
支持类：
//ABCOutput Class
@DefaultCoder(AvroCoder.class)
public class ABCOutput extends DoFn<KV<String, inResultPair>, KV<String, …java google-cloud-platform google-cloud-dataflow apache-beam
我正在尝试完成以下操作： Beam / Dataflow中的批处理PCollection
上面链接中的答案是Java，而我使用的语言是Python。因此，我需要一些帮助来获得类似的构造。
具体来说我有这个：
p = beam.Pipeline (options = pipeline_options)
lines = p | 'File reading' >> ReadFromText (known_args.input)
此后，由于我的用例需要一组行，因此我需要创建另一个，PCollection但要包含ListN行的“行”。我无法逐行操作。
我尝试了一个ParDo函数，该函数使用变量将计数与N行计数器相关联，并在groupBy使用之后Map。但是这些每1000条记录会重置一次，所以这不是我想要的解决方案。我阅读了链接中的示例，但我不知道如何在Python中执行类似的操作。
我尝试将计数器保存在数据存储区中，但是，使用数据存储区读写数据流之间的速度差异非常明显。
正确的方法是什么？我不知道该怎么办。问候。
我是Apache Beam的新手,我想计算大型数据集的平均值和标准偏差.
给定一个"A,B"形式的.csv文件,其中A,B是整数,这基本上就是我所拥有的.
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.textio import ReadFromText
class Split(beam.DoFn):
def process(self, element):
A, B = element.split(',')
return [('A', A), ('B', B)]
with beam.Pipeline(options=PipelineOptions()) as p:
# parse the rows
rows = (p
| ReadFromText('data.csv')
| beam.ParDo(Split()))
# calculate the mean
avgs = (rows
| beam.CombinePerKey(
beam.combiners.MeanCombineFn()))
# calculate the stdv per key
# ???
std >> beam.io.WriteToText('std.out')
我想做点什么:
class SquaredDiff(beam.DoFn):
def process(self, element):
A = element[0][1]
B = element[1][1]
return [('A', A …