标签: apache-beam

我正在关注这篇文章和文档的答案，以便在管道结束时对我的数据执行动态窗口写入。这是我到目前为止所拥有的：

static void applyWindowedWrite(PCollection<String> stream) {
    stream.apply(
        FileIO.<String, String>writeDynamic()
            .by(Event::getKey)
            .via(TextIO.sink())
            .to("gs://some_bucket/events/")
            .withNaming(key -> defaultNaming(key, ".json")));
}

但是 NetBeans 在最后一行警告我语法错误：

FileNaming is not public in Write; cannot be accessed outside package

如何使defaultNaming我的管道可用，以便我可以将其用于动态写入。或者，如果这不可能，我应该做什么？

我正在尝试通过 gcloud 命令运行数据流作业：

gcloud beta dataflow jobs run test --gcs-location gs://bucket/templates/templateName --parameters query="select a.name,b.salary,a.id from table1 a join table2 b on a.id = b.id"

但我收到一条错误消息：

错误：(gcloud.beta.dataflow.jobs.run) 参数 --parameters：dict arg 的错误语法：[b.salary]。请查看gcloud topic escaping您是否需要有关转义列表或字典标志值的信息。

我看到了 gcloud 主题转义的文档，但无法弄清楚如何在此处应用它。有人可以帮我解决这个问题。

谢谢。

我正在运行 Apache Beam 管道，从 Google Cloud Storage 读取文本文件，对这些文件执行一些解析并将解析后的数据写入 Bigquery。

为了保持简短，这里忽略解析和 google_cloud_options，我的代码如下：（apache-beam 2.5.0 with GCP add-ons and Dataflow as runner）

p = Pipeline(options=options)

lines = p | 'read from file' >> 
beam.io.ReadFromText('some_gcs_bucket_path*')  |  \
    'parse xml to dict' >> beam.ParDo(
        beam.io.WriteToBigQuery(
            'my_table',
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
    p.run()

这运行良好并成功地将相关数据附加到我的 Bigquery 表中以获取少量输入文件。但是，当我将输入文件的数量增加到 +- 800k 时，出现错误：

“BoundedSource.split() 操作返回的 BoundedSource 对象的总大小大于允许的限制。”

我发现故障排除 apache 光束管道导入错误 [BoundedSource 对象大于允许的限制]建议使用 ReadAllFromText 而不是 ReadFromText。
但是，当我换出时，出现以下错误：

p = Pipeline(options=options)

lines = p | 'read from file' >> 
beam.io.ReadFromText('some_gcs_bucket_path*')  |  \
    'parse …

我在 datalab 中使用 apache 光束/数据流“玩”。我正在尝试从 gcs 读取 csv 文件。当我使用以下方法创建 pcollection 时：

lines = p | 'ReadMyFile' >> beam.io.ReadFromText('gs://' + BUCKET_NAME + '/' + input_file, coder='StrUtf8Coder')

我收到以下错误：

LookupError: unknown encoding: "THE","NAME","OF","COLUMNS"

似乎列的名称被解释为编码？

我不明白出了什么问题。如果我不指定“编码器”，我会得到

UnicodeDecodeError: 'utf8' codec can't decode byte 0xe0 in position 1045: invalid continuation byte

在 apache 光束之外，我可以通过从 gcs 读取文件来处理此错误：

blob = storage.Blob(gs_path, bucket)
data = blob.download_as_string()
data.decode('utf-8', 'ignore')

我读到 apache beam 只支持 utf8 并且文件不只包含 utf8。

我应该下载然后转换为 pcollection 吗？

有什么建议吗？

我正在查看Apache Beam的 wordCount 示例 ，当我尝试在本地运行此示例时，它将计数写入多个文件。我创建了一个测试项目来从文件读取和写入数据，甚至该写入操作将输出写入多个文件。如何在单个文件中获得结果？我正在使用直接转轮

我的几个 PCollections（来自不同来源）必须以相同的方式解码。

hits = (msgs | 'Parse' >> beam.Map(parse)
    | 'Decode' >> beam.Map(decode_hit))

然后：

dummy_hits = (dummy_msgs | 'Parse' >> beam.Map(parse)
    | 'Decode' >> beam.Map(decode_hit))

多亏了我之前给它们的名字，如果我可以重用这些转换，那就太好了。我天真地尝试了这个：

dummy_hits = (dummy_msgs | 'Parse'
    | 'Decode')

但我的管道不会建立。（类型错误：需要一个 PTransform 对象，得到解析）。

我认为有可能作为管道模块的文档说明：“如果需要应用相同的转换实例，则应使用右移运算符来指定新名称（例如input | "label" >> my_tranform）”

这样做的方法是什么？只有这可能吗？

我正在构建 Apache Beam 管道，但在尝试导入管道选项时遇到了 AttributeError。

我正在使用 python3.6 在干净的虚拟环境中的 Ubuntu 服务器上进行测试

脚步：

virtualenv -p python3.6 beam-env
. beam-env/bin/activate
pip install apache_beam==2.12.0
python3.6 test.py

内部test.py：

from apache_beam.options.pipeline_options import PipelineOptions

我希望导入成功，但出现以下错误：

AttributeError: 模块 'apache_beam.coders.coders' 没有属性 'VarIntCoder'

我正在尝试在 Dataflow 中进行单元测试。

对于那个测试，在乞求时，我将从一个简单的硬编码字符串开始。

问题是我需要将该字符串转换为 pubsub 消息。我得到了以下代码来做到这一点：

    // Create a PCollection from string a transform to pubsub message format
    PCollection<PubsubMessage> input = p.apply("input string", Create.of("test" + 
            ""))
            .apply("convert to Pub/Sub message", ParDo.of(new DoFn<String, PubsubMessage>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                    c.output(new PubsubMessage(c.element().getBytes(), null));
                }
            }));

但我收到以下错误：

 java.lang.IllegalArgumentException: unable to serialize DoFnWithExecutionInformation{doFn=com.xxx.pipeline.TesterPipeline$1@7b64240d, mainOutputTag=Tag<output>, sideInputMapping={}, schemaInformation=DoFnSchemaInformation{elementConverters=[]}}
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:55)
    <...>
Caused by: java.io.NotSerializableException: com.xxx.pipeline.TesterPipeline
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) …

我的流程将文件存储在磁盘上，我需要设置文件名，以便我可以找回东西。

默认命名是窗口时间戳和计数器，这对我没有帮助。文档对我来说不够清楚。（https://beam.apache.org/releases/pydoc/2.16.0/apache_beam.io.fileio.html?highlight=default_file_naming）

fileio.WriteToFiles(archive_storage, file_naming=beam.io.fileio.destination_prefix_naming())

我想命名文件<HASH>.json中的 HASH 是文件中数据的文件。