标签: apache-beam

Apache Beam 最近通过注释引入了状态单元，并在Apache Flink 和 Google Cloud Dataflow 中提供了部分支持。StateSpec@StateId

我找不到任何有关将其与GlobalWindow. 特别是，有没有一种方法可以拥有“状态垃圾收集”机制，以根据某些配置摆脱一段时间未见过的键的状态，同时仍然保持已见过的键的单个历史状态足够频繁吗？

或者，在这种情况下使用的状态量是否会出现偏差，无法回收与一段时间未见过的键相对应的状态？

我还对 Apache Flink 或 Google Cloud Dataflow 是否支持潜在的解决方案感兴趣。

Flink 和 direct runner 似乎有一些“状态 GC”的代码，但我不太确定它的作用以及它在使用全局窗口时是否相关。

我正在编写一个数据流流管道。在其中一个转换中，DoFn 我想要访问外部服务 - 在本例中，它是数据存储区。

这种初始化步骤有没有最佳实践？我不想为每个 processElement 方法调用创建数据存储连接对象。

我正在尝试使用数据流读取 pubsub 消息并将其写入大查询。我获得了 Google 团队的 alpha 访问权限，并且已经使提供的示例正常工作，但现在我需要将其应用到我的场景中。

发布订阅负载：

Message {
    data: {'datetime': '2017-07-13T21:15:02Z', 'mac': 'FC:FC:48:AE:F6:94', 'status': 1}
    attributes: {}
}

大查询架构：

schema='mac:STRING, status:INTEGER, datetime:TIMESTAMP',

我的目标是简单地读取消息有效负载并插入到 bigquery 中。我正在努力了解转换以及如何将键/值映射到大查询模式。

我对此很陌生，因此非常感谢任何帮助。

当前代码： https: //codeshare.io/ayqX8w

谢谢！

我已经设置了管道。我必须解析数百个 *.gz 文件。因此 glob 效果很好。

但我需要当前处理的文件的原始名称，因为我想将结果文件命名为原始文件。

有人能帮我一下吗？

这是我的代码。

@Default.String(LOGS_PATH + "*.gz")
String getInputFile();
void setInputFile(String value);


    TextIO.Read read = TextIO.read().withCompressionType(TextIO.CompressionType.GZIP).from(options.getInputFile());
        read.getName();

        p.apply("ReadLines", read).apply(new CountWords())
         .apply(MapElements.via(new FormatAsTextFn()))
         .apply("WriteCounts", TextIO.write().to(WordCountOptions.LOGS_PATH + "_" + options.getOutput()));

    p.run().waitUntilFinish();

摘要：有些本地软件包可以工作，有些则不行

我的梁应用程序的结构：

-setup.py

-app/__init__.py
-app/main.py

-package1/__init__.py
-package1/one.py

-package2/__init__.py
-package2/two.py

-package3/__init__.py
-package3/three.py

在main.py中：

from package1 import one
from package2 import two
from package3 import three

在setup.py中

import setuptools

setuptools.setup(
    name='beam',
    version='1.0',
    install_requires=['apache-beam[gcp]',
                      'google-cloud==0.34.0',
                      'google-cloud-bigquery==0.25.0',
                      'requests==2.19.1',
                      'google-cloud-storage==1.12.0'
                      ],
    packages=setuptools.find_packages(),
)

运行时，通过使用python -m app.main：

使用直接运行程序（本地运行），没问题。

使用 DataflowRunner （发送到 gogole 数据流），我遇到此错误：

apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException：数据流管道失败。状态：失败，错误：回溯（最近一次调用）：文件“/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py”，第 642 行，在 do_work work_executor.execute() 文件中“ /usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py”，第 156 行，执行 op.start() 文件“apache_beam/runners/worker/operations.py”，第 344 行，在 apache_beam .runners.worker.operations.DoOperation.start def start(self)：文件“apache_beam/runners/worker/operations.py”，第 345 行，在 apache_beam.runners.worker.operations.DoOperation.start 中，带有 self.scoped_start_state：文件“apache_beam/runners/worker/operations.py”，第 350 行，在 apache_beam.runners.worker.operations.DoOperation.start pickler.loads(self.spec.serialized_fn)) 文件“/usr/local/lib/python2.7 /dist-packages/apache_beam/internal/pickler.py”，第 244 行，加载中返回 dill.loads(s) …

现在的情况

该管道的目的是从发布/订阅中读取带有地理数据的有效负载，然后对这些数据进行转换和分析，最后返回条件是否为真或假

 with beam.Pipeline(options=pipeline_options) as p:
        raw_data = (p
                    | 'Read from PubSub' >> beam.io.ReadFromPubSub(
                    subscription='projects/XXX/subscriptions/YYY'))

        geo_data = (raw_data
                    | 'Geo data transform' >> beam.Map(lambda s: GeoDataIngestion(s)))
                    
                    

def GeoDataIngestion(string_input):
    <...>
    return True or False

理想情况1

如果 GeoDataIngestion 结果为 true，则 raw_data 将存储在大查询中

geo_data = (raw_data
                | 'Geo data transform' >> beam.Map(lambda s: GeoDataIngestion(s))
                | 'Evaluate condition' >> beam.Map(lambda s: Condition(s))
                )

def Condition(condition):
    if condition:
        <...WriteToBigQuery...>


#The class I used before to store raw_data without depending on evaluate condition:

class WriteToBigQuery(beam.PTransform): …

我们通过气流使用阿帕奇光束。默认 GCS 帐户是使用环境变量 - GOOGLE_APPLICATION_CREDENTIALS 设置的。我们不想更改环境变量，因为它可能会影响当时运行的其他进程。我找不到以编程方式更改 Google Cloud Dataflow 服务帐户的方法。我们按照以下方式创建管道 p = beam.Pipeline(argv=self.conf)

是否有通过 argv 或选项的任何选项，我可以在其中提到 gcs 凭证文件的位置？搜索了文档，但没有找到太多信息。

我正在尝试使用 flex 模板运行我的 python 数据流作业。当我使用直接运行器（没有 flex 模板）运行时，作业在本地运行良好，但是当我尝试使用 flex 模板运行它时，作业卡在“排队”状态一段时间，然后因超时而失败。

这是我在 GCE 控制台中找到的一些日志：

INFO:apache_beam.runners.portability.stager:Executing command: ['/usr/local/bin/python', '-m', 'pip', 'download', '--dest', '/tmp/dataflow-requirements-cache', '-r', '/dataflow/template/requirements.txt', '--exists-action', 'i', '--no-binary', ':all:'

Shutting down the GCE instance, launcher-202011121540156428385273524285797, used for launching.

Timeout in polling result file: gs://my_bucket/staging/template_launches/2020-11-12_15_40_15-6428385273524285797/operation_result.
Possible causes are:
1. Your launch takes too long time to finish. Please check the logs on stackdriver.
2. Service my_service_account@developer.gserviceaccount.com may not have enough permissions to pull container image gcr.io/indigo-computer-272415/samples/dataflow/streaming-beam-py:latest or create new objects in gs://my_bucket/staging/template_launches/2020-11-12_15_40_15-6428385273524285797/operation_result.
3. Transient …

我试图将一个 Apache Beam 数据流管道更新到最新的 kotlin 1.4.30 版本，该管道当前正在 kotlin 1.4.21 上运行，但是一旦我将 build.gradle 升级到 1.4.30 版，编译就会失败，出现以下异常：

java.lang.IllegalStateException: Could not read class: VirtualFile: /Users/stefanomassera/.gradle/caches/modules-2/files-2.1/org.apache.beam/beam-sdks-java-core/2.26.0/b57f8fa5ae66564c7ffafde34b690057f471bfa8/beam-sdks-java-core-2.26.0.jar!/org/apache/beam/sdk/options/PipelineOptions.class
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.BinaryJavaClass.<init>(BinaryJavaClass.kt:120)
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.BinaryJavaClass.<init>(BinaryJavaClass.kt:34)
    at org.jetbrains.kotlin.cli.jvm.compiler.KotlinCliJavaFileManagerImpl.findClass(KotlinCliJavaFileManagerImpl.kt:115)
    at org.jetbrains.kotlin.resolve.jvm.KotlinJavaPsiFacade$CliFinder.findClass(KotlinJavaPsiFacade.java:484)
  ... omissis for brevity
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Wildcard mast have a bound for annotation of WILDCARD_BOUND position
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.BinaryJavaAnnotation$Companion.computeTargetType$resolution_common_jvm(Annotations.kt:188)
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.AnnotationsAndParameterCollectorMethodVisitor.visitTypeAnnotation$getTargetType(Annotations.kt:111)
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.AnnotationsAndParameterCollectorMethodVisitor.visitTypeAnnotation(Annotations.kt:117)
    at org.jetbrains.org.objectweb.asm.ClassReader.readMethod(ClassReader.java:1427)
    at org.jetbrains.org.objectweb.asm.ClassReader.accept(ClassReader.java:719)
    at org.jetbrains.org.objectweb.asm.ClassReader.accept(ClassReader.java:402)
    at org.jetbrains.kotlin.load.java.structure.impl.classFiles.BinaryJavaClass.<init>(BinaryJavaClass.kt:115)
    ... 101 more

我也尝试过一个空白的虚拟项目只是为了验证它是否与当前的 kotlin 和 apache beam 存在兼容性错误，这似乎是一个兼容性问题，这里是我的虚拟测试

主文件

package org.example

import org.apache.beam.sdk.Pipeline
import org.apache.beam.sdk.options.PipelineOptionsFactory

class DummyPipeline …

我每 6 分钟安排一个谷歌云数据流作业（使用 apache beam python sdk），它在内部从 Big Query Table 读取，进行一些转换并写入另一个 Big Query 表。此作业已开始间歇性失败（大约 10 次中的 4 次）并显示以下错误跟踪。

2021-02-17 14:51:18.146 ISTError message from worker: Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/dataflow_worker/batchworker.py", line 649, in do_work
    work_executor.execute()
  File "/usr/local/lib/python3.8/site-packages/dataflow_worker/executor.py", line 225, in execute
    self.response = self._perform_source_split_considering_api_limits(
  File "/usr/local/lib/python3.8/site-packages/dataflow_worker/executor.py", line 233, in _perform_source_split_considering_api_limits
    split_response = self._perform_source_split(source_operation_split_task,
  File "/usr/local/lib/python3.8/site-packages/dataflow_worker/executor.py", line 271, in _perform_source_split
    for split in source.split(desired_bundle_size):
  File "/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery.py", line 807, in split
    self.table_reference = self._execute_query(bq)
  File "/usr/local/lib/python3.8/site-packages/apache_beam/options/value_provider.py", line 135, in _f
    return …