标签: apache-beam

我有一个简单的 Google DataFlow 任务。它从 BigQuery 表中读取数据并写入另一个表，如下所示：

(p
 |  beam.io.Read( beam.io.BigQuerySource(
        query='select dia, import from DS1.t_27k where true', 
        use_standard_sql=True))
 |  beam.io.Write(beam.io.BigQuerySink(
                  output_table,
                  dataset='DS1', 
                  project=project, 
                  schema='dia:DATE, import:FLOAT',
                  create_disposition=CREATE_IF_NEEDED,
                      write_disposition=WRITE_TRUNCATE
                     )
                )

我想问题是这个管道似乎需要一个临时数据集才能完成工作。我无法强制该临时数据集的位置。因为我的 DS1 位于欧盟 (#EUROPE-WEST1)，而临时数据集位于美国（我猜），所以任务失败：

WARNING:root:Dataset m-h-0000:temp_dataset_e433a0ef19e64100000000000001a does not exist so we will create it as temporary with location=None
WARNING:root:A task failed with exception.
 HttpError accessing <https://www.googleapis.com/bigquery/v2/projects/m-h-000000/queries/b8b2f00000000000000002bed336369d?alt=json&maxResults=10000>: response: <{'status': '400', 'content-length': '292', 'x-xss-protection': '1; mode=block', 'x-content-type-options': 'nosniff', 'transfer-encoding': 'chunked', 'expires': 'Sat, 14 Oct 2017 20:29:15 GMT', 'vary': 'Origin, X-Origin', 'server': 'GSE', '-content-encoding': …

我正在将 Apache-Beam 与 Python SDK 结合使用。

目前，我的管道读取多个文件，解析它们并从其数据生成 pandas 数据帧。然后，它将它们分组到一个数据帧中。

我现在想要的是检索这个单一的胖数据帧，将其分配给一个普通的 Python 变量。

可以做吗？

输入PCollection是http请求，它是一个有界数据集。我想在 ParDo 中进行异步 http 调用（Java），解析响应并将结果放入输出 PCollection 中。我的代码如下。获取异常如下。

我不明白原因。需要指导....

java.util.concurrent.CompletionException: java.lang.IllegalStateException: Can't add element ValueInGlobalWindow{value=streaming.mapserver.backfill.EnrichedPoint@2c59e, pane=PaneInfo.NO_FIRING} to committed bundle in PCollection Call Map Server With Rate Throttle/ParMultiDo(ProcessRequests).output [PCollection]

代码：

public class ProcessRequestsFn extends DoFn<PreparedRequest,EnrichedPoint> {
    private static AsyncHttpClient _HttpClientAsync;
    private static ExecutorService _ExecutorService;

static{

    AsyncHttpClientConfig cg = config()
            .setKeepAlive(true)
            .setDisableHttpsEndpointIdentificationAlgorithm(true)
            .setUseInsecureTrustManager(true)
            .addRequestFilter(new RateLimitedThrottleRequestFilter(100,1000))
            .build();

    _HttpClientAsync = asyncHttpClient(cg);

    _ExecutorService = Executors.newCachedThreadPool();

}


@DoFn.ProcessElement
public void processElement(ProcessContext c) {

    PreparedRequest request = c.element();

    if(request == null)
        return;

    _HttpClientAsync.prepareGet((request.getRequest()))
            .execute()
            .toCompletableFuture()
            .thenApply(response -> …

我需要对数据流应用错误处理，以便使用相同的主键多次插入 Spanner。逻辑是，在当前消息之后可能会收到较旧的消息，并且我不想覆盖保存的值。因此，我将创建我的突变作为插入，并在尝试重复插入时抛出错误。

我在 DoFn 中看到过几个尝试块的示例，它们写入侧面输出以记录任何错误。这是一个非常好的解决方案，但我需要对写入不包含 DoFn 的 Spanner 的步骤应用错误处理

spannerBranchTuples2.get(spannerOutput2)
    .apply("Create Spanner Mutation", ParDo.of(createSpannerMutation))                      
    .apply("Write Spanner Records", SpannerIO.write()
        .withInstanceId(options.getSpannerInstanceId())                  
        .withDatabaseId(options.getSpannerDatabaseId())
        .grouped());

我还没有找到任何允许将错误处理应用于此步骤的文档，或者找到一种将其重写为 DoFn 的方法。有什么建议如何对此应用错误处理吗？谢谢

我正在尝试使用以下流程在 Google Dataflow 上运行作业：

本质上采用单个数据源，根据字典中的某些值进行过滤，并为每个过滤条件创建单独的输出。

我编写了以下代码：

# List of values to filter by
x_list = [1, 2, 3]

with beam.Pipeline(options=PipelineOptions().from_dictionary(pipeline_params)) as p:
    # Read in newline JSON data - each line is a dictionary
    log_data = (
        p 
        | "Create " + input_file >> beam.io.textio.ReadFromText(input_file)
        | "Load " + input_file >> beam.FlatMap(lambda x: json.loads(x))
    )
    
    # For each value in x_list, filter log_data for dictionaries containing the value & write out to separate file
    for i in x_list:
        # Return …

如何使用 Apache Beam Direct 运行程序通过 GOOGLE_APPLICATION_CREDENTIALS 进行身份验证？我不想使用 gcloud 帐户进行身份验证。我有一个服务帐户（json），我将其设置为系统变量。如何让 Apache Beam 程序（作为 DirectRunner 运行）使用 GOOGLE_APPLICATION_CREDENTIALS 进行身份验证？

我的用例是访问 Apache Beam 程序中的 GCP Pub/Sub 资源，因此需要进行身份验证

我们有一个有用户的应用程序；每个用户每次使用我们的应用程序大约 10-40 分钟，我想根据发生的特定事件（例如“此用户已转换”、“此用户上次会话出现问题”、“该用户上次会话成功”）。

（在此之后，我想计算每天这些更高级别的事件，但这是一个单独的问题）

为此，我一直在研究会话窗口；但所有文档似乎都面向全局会话窗口，但我想为每个用户创建它们（这也是自然分区）。

我无法找到有关如何执行此操作的文档（首选 python）。你能指出我正确的方向吗？

或者换句话说：如何创建可以输出更结构化（丰富）事件的每用户每会话窗口？

我拥有的

class DebugPrinter(beam.DoFn):
  """Just prints the element with logging"""
  def process(self, element, window=beam.DoFn.WindowParam):
    _, x = element
    logging.info(">>> Received %s %s with window=%s", x['jsonPayload']['value'], x['timestamp'], window)
    yield element

def sum_by_event_type(user_session_events):
  logging.debug("Received %i events: %s", len(user_session_events), user_session_events)
  d = {}
  for key, group in groupby(user_session_events, lambda e: e['jsonPayload']['value']):
    d[key] = len(list(group))
  logging.info("After counting: %s", d)
  return d

# ...

by_user = valid \
  | 'keyed_on_user_id'      >> beam.Map(lambda x: (x['jsonPayload']['userId'], x)) …

已经错过了窗口和晚期数据.withAllowedLateness周期从管道作为记录下车这里

我对这种行为有几个问题：

1. 如何处理从管道中丢失的后期数据？我们可以添加默认行为吗？说所有迟到的数据都应该记录在像包罗万象的桶之类的地方？
2. 我们能否有一个指标（Google Dataflow Metrics/Beam）来说明这些消息中有多少由于巨大的延迟而从管道中丢失？

我试图拥有我创建的 AutoValue 定义对象的 PCollection，并且我添加了适当的注释以通过DefaultSchema(AutoValueSchema.class). 就像这样：

@DefaultSchema(AutoValueSchema.class)
@AutoValue
public abstract class MyAutoClas {
  public abstract String getMyStr();
  public abstract Integer getMyInt();

  @CreateSchema
  public static MyAutoClass create(String myStr, Integer myInt) {
    return new AutoValue_MyAutoClass(myStr, myInt);
  }
}

我有一个小测试用例，如下所示：

PCollection<KV<String, MyAutoClass>> result = pipeline
    .apply(Create.of(MyAutoClass.create("abc", 1)))
    .apply(WithKeys.of(in -> in.getMyStr()));

PAssert.that(result).containsInAnyOrder(KV.of("abc", MyAutoClass.create("abc", 1)));
pipeline.run().waitUntilFinish();

当我尝试运行此程序时，我看到以下错误：

[ERROR] testMyAutoValueClass(.....)  Time elapsed: 1.891 s  <<< ERROR!
java.lang.RuntimeException: Creator parameter arg0 Doesn't correspond to a schema field
    at org.apache.beam.sdk.schemas.utils.ByteBuddyUtils$InvokeUserCreateInstruction.<init>(ByteBuddyUtils.java:717)
    at org.apache.beam.sdk.schemas.utils.ByteBuddyUtils$StaticFactoryMethodInstruction.<init>(ByteBuddyUtils.java:660)
    at org.apache.beam.sdk.schemas.utils.JavaBeanUtils.createStaticCreator(JavaBeanUtils.java:284)
    at org.apache.beam.sdk.schemas.utils.JavaBeanUtils.lambda$getStaticCreator$4(JavaBeanUtils.java:273)
    at …

我正在尝试将数据从外部 webhook/RSS 源流式传输到我的数据流中。我正在考虑使用发布/订阅来接收消息，然后在数据流中处理它。但是，我找不到这样做的选项。

除了设置侦听输入流的接收者队列服务器之外，是否有更好的方法在 GCP 作为托管服务中执行此操作？