当尝试运行此管道时，我收到此异常：

Exception has occurred: AttributeError module 'apache_beam.io.gcp.internal.clients.bigquery' has no attribute 'TableReference'

table_spec='ExporterPlayGround.TEST_STREAM' 
with beam.Pipeline(options=pipeline_options) as p:
        from apache_beam.io.gcp.internal.clients import bigquery
        raw_stream = (
            p | 'Start subscriber' >> beam.io.gcp.pubsub.ReadFromPubSub(subscription=subscription_name)
            | 'Write to Table' >> beam.io.WriteToBigQuery(
                table_spec,
                schema='test_float:FLOAT, test2_float:FLOAT',
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
            )

也许我错过了一些明显的东西，但我似乎看不出问题是什么，我正在提供文档指向的参考 bigquery。

Apache Beam 根据前一行的值更新值

我已将 CSV 文件中的值分组。在分组行中，我们发现一些缺失值，需要根据前一行的值进行更新。如果该行的第一列为空，那么我们需要将其更新为 0。

我能够对记录进行分组，但无法找出更新值的逻辑，我该如何实现这一点？

记录

分组记录

更新缺失值

到目前为止的代码：

public class GroupByTest {
    public static void main(String[] args) …

长话短说

我们有一个默认的 VPC。尝试运行数据流作业。初始步骤（读取文件）设法处理 1/2 步骤。获取JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected错误消息，但日志中没有其他内容。已尝试设置角色和 vpc 防火墙规则。

问题

我想使用 Geobeam 图像 (Apache Beam Python 3.9 SDK 2.41.0) 运行数据流作业。我对工作的定义如下：

def run(pipeline_args, known_args):
    import apache_beam as beam
    from apache_beam.io.gcp.internal.clients import storage
    from apache_beam.options.pipeline_options import PipelineOptions
    from geobeam.io import GeoJSONSource, filebasedsource
    from geobeam.fn import format_record, make_valid, filter_invalid

    pipeline_options = PipelineOptions([
        
    ] + pipeline_args)

    with beam.Pipeline(options=pipeline_options) as p:
        (p
         | beam.io.Read(GeoJSONSource(known_args.gcs_url, encoding='utf-8'))
         | 'FilterCords' >> beam.Filter(lambda x: len(x[-1]["coordinates"]) > 1)
         | 'MakeValid' >> beam.Map(make_valid)
         | 'FilterInvalid' >> beam.Filter(filter_invalid) …

我们正在Google Cloud上建立机器学习管道,利用GC ML-Engine进行分布式TensorFlow培训和模型服务,并利用DataFlow进行分布式预处理作业.

我们希望在Google Cloud上运行我们的Apache Beam应用程序作为DataFlow作业.看看ML-Engine样本 ,似乎可以得到tensorflow_transform.beam.impl AnalyzeAndTransformDataset来指定使用哪个PipelineRunner,如下所示:

from tensorflow_transform.beam import impl as tft
pipeline_name = "DirectRunner"
p = beam.Pipeline(pipeline_name) 
p | "xxx" >> xxx | "yyy" >> yyy | tft.AnalyzeAndTransformDataset(...)

TemplatingDataflowPipelineRunner提供了将预处理开发与参数化操作分开的功能 - 请参阅此处:https://cloud.google.com/dataflow/docs/templates/overview-基本上:

• A)在PipelineOptions派生类型中,将选项类型更改为ValueProvider(python方式:类型推断或类型提示???)
• B)将跑步者改为TemplatingDataflowPipelineRunner
• C) mvn原型:生成以GCS存储模板(python方式:像TF Hypertune一样的yaml文件???)
• D)运行gcloud beta数据流作业--gcs-location -parameters

现在的问题是:你能告诉我,我们怎能用tf.Transform利用TemplatingDataflowPipelineRunner？

我正在尝试用Python编写一个数据流管道，该管道需要一个大的numpy矩阵作为辅助输入。矩阵保存在云存储中。理想情况下，每个Dataflow工作人员都可以直接从云存储中加载矩阵。

我的理解是，如果我说matrix = np.load(LOCAL_PATH_TO_MATRIX)，然后

p | "computation" >> beam.Map(computation, matrix)

矩阵从我的笔记本电脑运送到每个Datflow工作人员。

我该如何指导每个工作人员直接从云存储中加载矩阵？有用于“二进制斑点”的光束源吗？

我们如何通过Google数据流作业在GCS中创建压缩文件？

我无法指定压缩类型.如果该功能尚未出现,是否有更简洁的方法从Google BigQuery的查询输出到压缩文件？

如果我的类扩展了DoFn,如何访问侧输入的元素？

例如:

假设我有一个ParDo变换,如:

PCollection<String> data = myData.apply("Get data",
    ParDo.of(new MyClass()).withSideInputs(myDataView));

我有一节课: -

static class MyClass extends DoFn<String,String>
{
    //How to access side input here
}

在这种情况下,c.sideInput()不起作用.

谢谢.

我的流水线的第一步涉及从外部数据源中获取数据，我想分块进行（顺序无关紧要）。我找不到任何类似的类，因此创建了以下内容：

class FixedSizeBatchSplitter(beam.DoFn):
  def __init__(self, size):
    self.size = size

  def start_bundle(self):
    self.current_batch = []

  def finish_bundle(self):
    if self.current_batch: 
      yield self.current_batch

  def process(self, element):
    self.current_batch.append(element)
    if len(self.current_batch) >= self.size:
      yield self.current_batch
      self.current_batch = []

但是，当我运行此管道时，出现RuntimeError: Finish Bundle should only output WindowedValue type错误：

with beam.Pipeline() as p:
  res = (p
         | beam.Create(range(10))
         | beam.ParDo(FixedSizeBatchSplitter(3))
        )

这是为什么？我怎么才能在process而不是在其中产生输出finish_bundle？顺便说一句，如果我删除finish_bundle管道的工作，但显然会丢弃剩余的。

由于文档仅适用于JAVA,我无法理解其含义.

它声明 - "虽然ParDo总是产生一个主输出PCollection(作为应用的返回值),你也可以让你的ParDo产生任意数量的额外输出PCollections.如果你选择有多个输出,你的ParDo将返回所有的输出PCollections(包括主输出)捆绑在一起.例如,在Java中,输出PCollections捆绑在一个类型安全的PCollectionTuple中."

我理解捆绑在一起意味着什么,但如果我在我的DoFn中产生一个标签,它是否会产生一个包含所有其他输出的空包,并在代码中遇到它们时产生其他输出？或者它等待所有产量准备好输入并将它们全部输出到一起？

文档中没有太多清晰度.虽然我认为它不会等待,只是遇到收益,但我仍然需要了解发生了什么.

我有一个GCS存储桶，我正试图从其中读取约200k个文件，然后将它们写入BigQuery。问题是我在创建与代码配合良好的PCollection时遇到了麻烦。我正在按照本教程进行参考。

我有以下代码：

from __future__ import absolute_import

import argparse
import logging
import os

from past.builtins import unicode

import apache_beam as beam
from apache_beam.io import ReadFromText, ReadAllFromText
from apache_beam.io import WriteToText
from apache_beam.metrics import Metrics
from apache_beam.metrics.metric import MetricsFilter
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from google.cloud import storage

import regex as re

# storage_client = storage.Client()
# bucket = storage_client.get_bucket('mybucket')
#
# blobs = bucket.list_blobs()
# l=list(blobs)
# x=[y.name for y in l]
# c=x[1:]
# print(len(c)) …