标签: apache-beam

我有以下 BigTable 结构作为示例：

Table1 : column_family_1 : column_1 : value

这value是一个数字。这是由数据流管理的，我想每次都更新该值。

该值可能是一个金额，我想在每次用户购买时更新它（以维持迄今为止的总支出），因此我在购买事件侦听器数据流中执行以下操作（每当遇到购买事件时）：

• 发出 BigTable 请求，通过 id 获取值
• 将新购买的金额添加到 BigTable 搜索响应中显示的金额中
• 发出Put更新值的请求

尽管这种方法有一些网络延迟，但它似乎有效。失败的情况是，当数据流有多个工作人员时，用户进行多次购买，并且事件会发送给多个工作人员，例如：

• Worker 1 获取事件 1，获取金额并将花费的金额添加到其中
• Worker 2 获取事件 2，获取旧金额并将花费的金额添加到其中
• 两个工人都提出Put请求，但他们都被覆盖了

为了防止这种情况，我试图提出一个仅以纯文本形式表示的请求，add 10 to the spent amount value. 这是我们可以在数据流中做的事情吗？

目前使用 Google Dataflow 和 Python 进行批处理。这工作得很好，但是，我有兴趣在不处理 Java 的情况下提高我的数据流作业的速度。

使用 Go SDK，我实现了一个简单的管道，它从 Google Storage读取一系列100-500mbtextio.Read文件（使用），进行一些聚合并使用结果更新 CloudSQL。正在读取的文件数量可以从数十个到数百个不等。

当我运行管道时，我可以从日志中看到文件是串行读取的，而不是并行读取的，因此作业需要更长的时间。使用 Python SDK 执行的同一进程会触发自动缩放并在几分钟内运行多次读取。

我尝试使用 指定工作人员数量--num_workers=，但是，Dataflow 在几分钟后将作业缩减为一个实例，并且从日志来看，在实例运行期间不会发生并行读取。

如果我删除textio.Read并实现自定义 DoFn 以从 GCS 读取，则会发生类似的情况。读取过程仍然串行运行。

我知道当前的 Go SDK 是实验性的，缺乏许多功能，但是，我还没有找到对并行处理限制的直接参考，在这里。Go SDK 的当前版本是否支持 Dataflow 上的并行处理？

提前致谢

我正在尝试从 Apache Beam 读取 Pubsub 消息的时间戳值。

p.apply("Read PubSub messages", PubsubIO.readMessagesWithAttributes()
    .withIdAttribute("msg_id")
    .withTimestampAttribute("timestamp")
    .fromSubscription(options.getPubsubSubscription()))

但不幸的是，我收到了以下错误，这真的让我感到惊讶，因为我认为每条消息都有一个默认时间戳。

An exception occured while executing the Java class. 
PubSub message is missing a value for timestamp attribute timestamp

为什么我的消息没有时间戳？是因为我通过 Pubsub UI 发布了它吗？

我正在尝试使用开源框架进行 ETL，我听说过两种东西 Apache Beam 和 Apache Airflow，其中一种最适合整个 ETL 或 ELT，例如 Talend、Azure 数据工厂等，事实上，我尝试使用云数据仓库（redshift、azure 数据仓库、雪花等）完成所有工作，哪一个适合此类工作，如果我对这两个框架进行一些比较，那就太好了。提前致谢。

Apache Beam 2.12.0 是否支持 Java 11，还是我现在应该继续使用稳定的 Java 8 SDK？

我看到该站点根据文档推荐使用 Python 3.5 和 Beam 2.12.0，与其他更高的 Python 版本相比。这个时候它和 Java 11 有多少可比性。因此，与 Apache Beam 2.12.0 一起使用的稳定版本仍然是 Java 8。在 Java 11 中使用 Beam 2.12.0 时，我遇到的构建问题很少。

最近在学习apache beam，发现了一些像这样的python代码：

lines = p | 'read' >> ReadFromText(known_args.input)

  # Count the occurrences of each word.
  def count_ones(word_ones):
    (word, ones) = word_ones
    return (word, sum(ones))

  counts = (lines
            | 'split' >> (beam.ParDo(WordExtractingDoFn())
                          .with_output_types(unicode))
            | 'pair_with_one' >> beam.Map(lambda x: (x, 1))
            | 'group' >> beam.GroupByKey()
            | 'count' >> beam.Map(count_ones))

来自： https: //github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount.py#L92

|python中and的语法和用法是什么>>？

我正在尝试在 GCP Dataflow 上运行一个简单的 Beam 脚本，以便将 scikit-learn 模型应用于某些数据。应用模型之前和之后都需要对数据进行处理。这就是textExtraction 和translationDictionary 的作用。我不断收到错误AttributeError: module 'google.cloud' has no attribute 'storage'（下面是完整的堆栈跟踪）。正如您所看到的，我尝试在新的虚拟环境中运行新的安装。知道如何修复吗？

我的脚本也在下面给出。

预测_DF_class.py

import apache_beam as beam
import argparse
from google.cloud import storage
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
import pandas as pd
import pickle as pkl
import json
import joblib

query = """
    SELECT index, product_title
    FROM `project.dataset.table`
    LIMIT 1000
"""

class ApplyDoFn(beam.DoFn):
    def __init__(self):
        self._model = None
        self._textExtraction = None
        self._translationDictionary = None
        self._storage = …

我们在 Elastic Cloud 上托管 elatsicsearch 集群并从数据流 (GCP) 调用它。工作在开发中运行良好，但当我们部署到产品时，我们在客户端看到大量连接超时。

Traceback (most recent call last):
  File "apache_beam/runners/common.py", line 1213, in apache_beam.runners.common.DoFnRunner.process
  File "apache_beam/runners/common.py", line 570, in apache_beam.runners.common.SimpleInvoker.invoke_process
  File "main.py", line 159, in process
  File "/usr/local/lib/python3.7/site-packages/elasticsearch/client/utils.py", line 152, in _wrapped
    return func(*args, params=params, headers=headers, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/elasticsearch/client/__init__.py", line 1617, in search
    body=body,
  File "/usr/local/lib/python3.7/site-packages/elasticsearch/transport.py", line 390, in perform_request
    raise e
  File "/usr/local/lib/python3.7/site-packages/elasticsearch/transport.py", line 365, in perform_request
    timeout=timeout,
  File "/usr/local/lib/python3.7/site-packages/elasticsearch/connection/http_urllib3.py", line 258, in perform_request
    raise ConnectionError("N/A", str(e), e)
elasticsearch.exceptions.ConnectionError: ConnectionError(<urllib3.connection.HTTPSConnection object at 0x7fe5d04e5690>: Failed …

我想估算在 Google Cloud DataFlow 上运行的 Apache Beam Pipeline 成本，我目前正在运行 Apache Beam 代码，该代码使用 for 循环自动扩展管道，并在聚合和处理之前将数据存储在 Beam 本身中 12 小时左右管道中的数据。任何关于如何估计成本的想法都将受到赞赏，并且也将有优化方法来最小化这种成本。

谢谢！

我正在编写一个数据流作业，它从 BigQuery 读取数据并进行一些转换。

data = (
    pipeline
    | beam.io.ReadFromBigQuery(query='''
    SELECT * FROM `bigquery-public-data.chicago_crime.crime` LIMIT 100
    ''', use_standard_sql=True)
    | beam.Map(print)
)

但我的要求是仅在收到来自 PubSub 主题的通知后才从 BigQuery 读取。仅当收到以下消息时，上述 DataFlow 作业才应开始从 BigQuery 读取数据。如果是不同的作业 ID 或不同的状态，则不应执行任何操作。

PubSub Message : {'job_id':101, 'status': 'Success'}

这部分有什么帮助吗？

在 GCP Dataflow 上运行的 Apache Beam Python SDK 中，我需要DoFn.process很长时间。我的 DoFn 花了很长时间，原因并不那么重要 - 由于我无法控制的要求，我必须接受它们。但如果您必须知道的话，它对外部服务进行网络调用需要相当长的时间（几秒钟），并且它正在处理先前的多个元素GroupByKey- 导致DoFn.process调用需要几分钟的时间。

无论如何，我的问题是：通话的运行时间长度是否有时间限制DoFn.process？我这么问是因为我看到的日志如下所示：

WARNING 2023-01-03T13:12:12.679957Z ReportProgress() took long: 1m49.15726646s
WARNING 2023-01-03T13:12:14.474585Z ReportProgress() took long: 1m7.166061638s
WARNING 2023-01-03T13:12:14.864634Z ReportProgress() took long: 1m58.479671042s
WARNING 2023-01-03T13:12:16.967743Z ReportProgress() took long: 1m40.379289919s
2023-01-03 08:16:47.888 EST Error message from worker: SDK harness sdk-0-6 disconnected.
2023-01-03 08:21:25.826 EST Error message from worker: SDK harness sdk-0-2 disconnected.
2023-01-03 08:21:36.011 EST Error message from worker: SDK harness sdk-0-4 disconnected. …

我想知道应该在 GCP pubsub 中创建什么类型的订阅，以便处理来自 pubsub 主题的高频数据。我将以每秒 100 多条消息的速度在数据流中摄取数据。拉或推订阅真的很重要，以及它将如何影响速度等等。