根据Apache Beam 2.0.0 SDK 文档 GroupIntoBatches仅适用于KV集合。
我的数据集只包含值，不需要引入键。然而，为了使用GroupIntoBatches我必须用一个空字符串作为键来实现“假”键：
static class FakeKVFn extends DoFn<String, KV<String, String>> {
@ProcessElement
public void processElement(ProcessContext c) {
c.output(KV.of("", c.element()));
}
}
所以整体管道如下所示：
public static void main(String[] args) {
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
long batchSize = 100L;
p.apply("ReadLines", TextIO.read().from("./input.txt"))
.apply("FakeKV", ParDo.of(new FakeKVFn()))
.apply(GroupIntoBatches.<String, String>ofSize(batchSize))
.setCoder(KvCoder.of(StringUtf8Coder.of(), IterableCoder.of(StringUtf8Coder.of())))
.apply(ParDo.of(new DoFn<KV<String, Iterable<String>>, String>() {
@ProcessElement
public void processElement(ProcessContext c) {
c.output(callWebService(c.element().getValue()));
}
}))
.apply("WriteResults", TextIO.write().to("./output/"));
p.run().waitUntilFinish();
}
有没有办法在不引入“假”密钥的情况下分组？
如何使用 Apache Beam 中的 TextIO 将从 PubSub 收到的消息写入 GCS 中的文本文件？看到了一些方法，如 withWindowedWrites() 和 withFilenamePolicy() 但在文档中找不到任何示例。
假设我们有一个有 4 个 CPU 内核的工人。Dataflow 工作机器中的并行性如何配置？我们是否并行超过内核数量？
在哪里可以获得此类信息？
我有一个存储为模板的管道。我正在使用 node.js 客户端从云函数运行此管道。一切正常，但是当我需要从不同区域运行此模板时，我会出错。
根据文档，我可以通过payload中的location参数进行设置
{
projectId: 123,
resource: {
location: "europe-west1",
jobName: `xxx`,
gcsPath: 'gs://xxx'
}
}
这给了我以下错误：
The workflow could not be created, since it was sent to an invalid regional endpoint (europe-west1).
Please resubmit to a valid Cloud Dataflow regional endpoint.
如果我将位置参数移出资源节点，我会得到同样的错误，例如：
{
projectId: 123,
location: "europe-west1",
resource: {
jobName: `xxx`,
gcsPath: 'gs://xxx'
}
}
如果我在环境中设置区域并删除位置，例如：
{
projectId: 123,
resource: {
jobName: `xxx`,
gcsPath: 'gs://xxx',
environment: {
zone: "europe-west1-b"
}
}
}
我不再收到任何错误，但数据流 UI 告诉我作业正在运行 us-east1
如何运行此模板并提供区域/区域 I
从：
https://github.com/spotify/scio/wiki/Scio-data-guideline
“比 groupByKey 更喜欢组合/聚合/减少转换。请记住，减少操作必须是关联的和可交换的。”
为什么特别喜欢聚合而不是 groupByKey？
我已经按照本答案中的描述实现了N个元素的批处理: google dataflow管道中的数据存储输入是否可以一次处理一批N个条目？
package com.example.dataflow.transform;
import com.example.dataflow.event.ClickEvent;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.joda.time.Instant;
import java.util.ArrayList;
import java.util.List;
public class ClickToClicksPack extends DoFn> {
public static final int BATCH_SIZE = 10;
private List accumulator;
@StartBundle
public void startBundle() {
accumulator = new ArrayList(BATCH_SIZE);
}
@ProcessElement
public void processElement(ProcessContext c) {
ClickEvent clickEvent = c.element();
accumulator.add(clickEvent);
if (accumulator.size() >= BATCH_SIZE) {
c.output(accumulator);
accumulator = new ArrayList(BATCH_SIZE);
}
}
@FinishBundle
public void finishBundle(FinishBundleContext c) {
if (accumulator.size() > 0) {
ClickEvent clickEvent = accumulator.get(0); …Run Code Online (Sandbox Code Playgroud) 我正在通过ReadFromPubSubwith阅读消息timestamp_attribute=None，它应该将时间戳设置为发布时间。
这样，我最终得到 a PCollectionofPubsubMessage元素。
如何按顺序访问这些元素的时间戳，例如将它们保存到数据库？我能看到的唯一属性是dataand attributes，并且attributes只有来自 Pub/Sub 的键。
编辑：示例代码
with beam.Pipeline(options=pipeline_options) as p:
items = (p
| ReadFromPubSub(topic=args.read_topic, with_attributes=True)
| beam.WindowInto(beam.window.FixedWindows(args.time_window))
| 'FormatMessage' >> beam.Map(format_message)
| 'WriteRaw' >> WriteToBigQuery(args.raw_table, args.dataset,
args.project, write_disposition='WRITE_APPEND')
)
whereformat_message将使用 aPubsubMessage并返回一个字典，表示要附加到表中的行：
def format_message(message):
formatted_message = {
'data': base64.b64encode(message.data),
'attributes': str(message.attributes)
}
return formatted_message
我正在实施 Pub/Sub 到 BigQuery 管道。它看起来类似于How to create read transform using ParDo and DoFn in Apache Beam，但在这里，我已经创建了一个 PCollection。
我正在遵循Apache Beam 文档中描述的内容来实现 ParDo 操作以使用以下管道准备表行：
static class convertToTableRowFn extends DoFn<PubsubMessage, TableRow> {
@ProcessElement
public void processElement(ProcessContext c) {
PubsubMessage message = c.element();
// Retrieve data from message
String rawData = message.getData();
Instant timestamp = new Instant(new Date());
// Prepare TableRow
TableRow row = new TableRow().set("message", rawData).set("ts_reception", timestamp);
c.output(row);
}
}
// Read input from Pub/Sub
pipeline.apply("Read from …python -m main \ --setup_file setup.py \ --runner DataflowRunner \ --project my-test \ --staging_location gs://my-test/staging \ --temp_location gs://my-test/temp \ --template_location gs://my-test/templates/test --output gs://my-test/output
上面的命令只在本地运行（本地安装的需要依赖项）并且不创建模板。这是 main.py 中的管道选项：
pipeline_options = {
'project': 'my-test',
'staging_location': 'gs://my-test/staging',
'runner': 'DataflowRunner',
'job_name': 'test',
'temp_location': 'gs://my-test/temp',
'save_main_session': True,
'setup_file':'setup.py',
'output': 'gs://my-test/output',
'template_location': 'gs://my-test/templates/test'
}
options = PipelineOptions.from_dictionary(pipeline_options)
with beam.Pipeline(options=options) as p:
这是 setup.py：
import subprocess
import setuptools
from setuptools.command.bdist_egg import bdist_egg as _bdist_egg
class bdist_egg(_bdist_egg):
def run(self):
self.run_command('CustomCommands')
_bdist_egg.run(self)
CUSTOM_COMMANDS = [
['apt-get', 'update'],
['apt-get', '--assume-yes', 'install', …TypeError: 'PCollection' object does not support indexing
上述错误是由于尝试将 Pcollection 转换为列表而导致的：
filesList = (files | beam.combiners.ToList())
lines = (p | 'read' >> beam.Create(ReadSHP().ReadSHP(filesList))
| 'map' >> beam.Map(_to_dictionary))
和：
def ReadSHP(self, filesList):
"""
"""
sf = shp.Reader(shp=filesList[1], dbf=filesList[2])
如何解决这个问题？任何帮助表示赞赏。