如何正确使用数据流/ Apache beam wait_until_finish持续时间参数？
数据流/ Apache Beam中的wait_until_finish持续时间参数用于控制作业的运行时间。它指定了作业在没有新数据输入时应该等待多长时间，然后才能完成作业。
在使用wait_until_finish持续时间参数时，需要考虑以下几点：
- 数据流作业的特性：根据数据流作业的特性和需求，合理设置wait_until_finish持续时间参数。如果作业需要实时处理数据，可以将持续时间设置为较短的时间间隔，以便及时响应新数据的到达。如果作业是批处理作业，可以将持续时间设置为较长的时间间隔，以便充分利用资源进行计算。
- 数据流的延迟和吞吐量：考虑数据流的延迟和吞吐量对作业运行时间的影响。如果数据流的延迟较高，可以适当增加持续时间参数，以确保作业能够处理到达的所有数据。如果数据流的吞吐量较大，可以适当减少持续时间参数，以避免作业过长的运行时间。
- 作业的稳定性和可靠性：根据作业的稳定性和可靠性要求，设置合适的持续时间参数。如果作业需要保证结果的准确性和完整性，可以将持续时间设置为较长的时间间隔，以确保所有数据都被处理。如果作业对结果的实时性要求不高，可以将持续时间设置为较短的时间间隔，以提高作业的响应速度。
总结起来，正确使用数据流/ Apache Beam的wait_until_finish持续时间参数需要根据作业的特性、数据流的延迟和吞吐量、作业的稳定性和可靠性等因素进行综合考虑。合理设置持续时间参数可以提高作业的效率和性能。
腾讯云相关产品和产品介绍链接地址：
- 数据流处理：腾讯云流计算 TDS：https://cloud.tencent.com/product/tds
- 批量数据处理：腾讯云数据处理 DCP：https://cloud.tencent.com/product/dcp
- 数据仓库：腾讯云数据仓库 CDW：https://cloud.tencent.com/product/cdw
- 数据库：腾讯云数据库 TencentDB：https://cloud.tencent.com/product/tencentdb
- 人工智能：腾讯云人工智能 AI：https://cloud.tencent.com/product/ai
- 物联网：腾讯云物联网 IoV：https://cloud.tencent.com/product/iov
- 移动开发：腾讯云移动开发 MDP：https://cloud.tencent.com/product/mdp
- 存储：腾讯云对象存储 COS：https://cloud.tencent.com/product/cos
- 区块链：腾讯云区块链 TBaaS：https://cloud.tencent.com/product/tbaas
- 元宇宙：腾讯云元宇宙 TME：https://cloud.tencent.com/product/tme
相关·内容
我正在尝试使用python code.My代码中的kafka.ReadFromKafka()方法读取kafka主题中的数据，如下所示：
from apache_beam.io.external import kafka
import apache_beam as beam
options = PipelineOptions()
with beam.Pipeline(options=options) as p:
plants = (
p
| 'read' >> kafka.ReadFromKafka({
浏览 1提问于2020-07-07得票数 0
我有一个批处理作业在数据流运行器上的版本apache-beamgcp==2.19.0下的gcp上的数据流中运行。我为该作业创建了一个自定义模板。作业按预期运行，但我还想添加一个最大作业持续时间。I found the duration (in milliseconds) parameter inside the wait_until_finish() method，which should be available.问题是:如何让模板化的批处理作业在运行时间超过持续时间时自动停止？我不需要保留任何数据，我只想让作业在运行时间过长时停止。我已经实现了run函数，如下所示： def run():
浏览 22提问于2020-08-13得票数 0
我已经建立了一个DAG来运行一个数据流作业。Dag很好地触发它，并且它成功地运行，但是输出文件没有出现在输出位置。输出位置是另一个项目中的一个桶，所使用的SA有权写入该桶.知道为什么文件没有生成吗？
DF职务：
import apache_beam as beam
from apache_beam.options.value_provider import StaticValueProvider
from apache_beam.options.pipeline_options import PipelineOptions
from datetime import datetime
impor
浏览 2提问于2022-02-21得票数 0
我正在尝试使用Python从GCP上的云函数执行数据流管道。在使用DataFlowRunner的笔记本服务器上测试代码。但是，当使用云函数调用管道时，我得到以下信息：
错误
Traceback (most recent call last): File "/env/local/lib/python3.7/site-packages/google/cloud/functions/worker.py",
line 346, in run_http_function result = _function_handler.invoke_user_function(flask.req
浏览 2提问于2020-03-10得票数 0
回答已采纳
1回答
我实现了一个具有多个无界源和侧输入的数据管道，用滑动窗口连接数据(30s和10s)，并将转换后的输出发送到Kafka主题中。我的问题是，在窗口的前10秒内收到的数据被发出了3次(即)。每当新窗口开始时，直到第一个窗口完成为止，都会触发。如何只发出一次转换后的数据或避免重复？
我使用过废弃的烧制窗格，这并没有什么区别。每当我尝试将窗口关闭行为设置为FIRE_ALWAYS/FIRE_NON时，它会抛出以下错误。
线程"main“org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumen
浏览 0提问于2019-07-17得票数 0
我正在尝试设置一个数据流作业，用于将json文件转换为csv，并使用下面的python脚本将其写入桶中。(我在3.8.13中尝试了这一点)，因为我使用的是apache。我尝试过改变许多版本的python和google云存储。在不使用存储库的情况下，有什么替代方法吗？
import apache_beam as beam
import pandas as pd
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import storage
from smart_open impor
浏览 6提问于2022-07-06得票数 0
回答已采纳
我正在使用一个自定义的worker_harness_container_image (和experiment=beam_fn_api)在数据流上尝试一个非常简单的管道： main.py import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, SetupOptions
import logging
def run(argv=None, save_main_session=True):
parser = argparse.A
浏览 30提问于2020-01-21得票数 0
我正在开发一个python程序，以像Google模板一样使用。
我正在做的是用BigQuery从PubSub编写数据：
pipeline_options.view_as(StandardOptions).streaming = True
p = beam.Pipeline(options=pipeline_options)
(p
# This is the source of the pipeline.
| 'Read from PubSub' >> beam.io.ReadFromPubSub('projects/.
浏览 0提问于2018-11-05得票数 0
回答已采纳
我想将BigQuery表名作为运行时参数传递给我的数据流模板，就像这样： class UserOptions(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument('--input', type=str, help='BigQuery table reference DATASET.TABLE')
parser.add_value_provider_argumen
浏览 12提问于2020-12-16得票数 0
我正在尝试模仿这个演练here，以便构建一个管道来将sklearn模型应用于我拥有的数据。我的命令行输入和随之而来的错误如下： (venv) computer:predictions uswygst$ python predictions.py \
--runner DataflowRunner \
--project my_project \
--requirements_file "requirements.txt" \
--temp_location gs://my_bucket/template/ \
--worker_machine_type n1
浏览 9提问于2020-09-16得票数 0
回答已采纳
我试图运行一个简单的beam脚本上的GCP数据流，以便应用一个科学工具包-学习模型的一些数据。数据需要在应用模型之前和之后进行处理。这就是textExtraction和translationDictionary的特点。我一直在获取错误AttributeError: module 'google.cloud' has no attribute 'storage' (下面是完整的堆栈跟踪)。如您所见，我尝试使用新的安装在一个新的虚拟环境中运行。知道怎么修吗？
下面我也给出了我的脚本。
predict_DF_class.py
import apache_beam as
浏览 12提问于2020-09-13得票数 0
回答已采纳
我试图从GCS桶中读取XML文件的集合，并处理它们，其中集合中的每个元素都是一个字符串，表示整个文件，但我找不到一个关于如何实现这一点的好例子，也无法从主要关于Java版本的Apache文档中理解它。
我目前的管道如下：
p = beam.Pipeline(options=PipelineOptions(pipeline_args))
(p
| 'Read from a File' >> beam.io.Read(training_files_folder)
| 'String To BigQuery Row' >> beam.Ma
浏览 6提问于2018-08-22得票数 3
回答已采纳
试图部署一条管道。我正在遵循以下实现：
尽管稍微改变了它，因为映射数据在csv文件中，而不是在bq中。
错误消息：
File "dataflow_pipeline.py", line 193, in <module>
run()
File "dataflow_pipeline.py", line 188, in run
p.run().wait_until_finish()
File "/home/demirfaruk/.local/lib/python3.7/site-packages/apache_beam/run
浏览 4提问于2020-12-06得票数 0
回答已采纳
我目前正在使用Dataflow来处理/按摩来自PubSub的XML文本字符串。我成功地使用DirectRunner为我的-runner标志运行了一个数据流作业。但是，我遇到了一个问题，试图使用与DataflowRunner完全相同的数据流作业来创建数据流资源。
从错误日志(在使用DataflowRunner时)来看，我创建的Dataflow模板似乎不识别：
import xml.etree.ElementTree as ET
每当我在管道中引用ET时，我都会收到"NameError: name‘ET“在运行’generatedPTrans-419‘时没有定义。奇怪的是，我的数据流工作在
浏览 1提问于2020-07-07得票数 0
我想得到运行在Google上的Apache管道成本的估算，我目前正在运行一个Apache代码，它使用for循环自动缩放管道，并将数据存储在Beam中12小时左右，然后聚合和处理管道中的数据。任何关于我如何估算成本的想法都会受到赞赏，并且优化方法，以尽量减少这种成本。
谢谢!
浏览 13提问于2022-10-13得票数 0
回答已采纳
我是GCP数据流的新手。
我尝试将文本文件(一行JSON字符串)从GCP云存储中读取为JSON格式，然后根据特定字段的值将其拆分并输出到GCP云存储(作为JSON字符串文本文件)。
这是
但是，我在GCP数据流上遇到了一些错误：
Traceback (most recent call last):
File "main.py", line 169, in <module>
run()
File "main.py", line 163, in run
shard_name_template='')
File
浏览 6提问于2019-10-16得票数 5
1回答
我用Bigquery编写了一个存储过程，并试图在数据流管道中调用它。这适用于SELECT查询，但不适用于存储过程：
pipeLine = beam.Pipeline(options=options)
rawdata = ( pipeLine
| beam.io.ReadFromBigQuery(
query="CALL my_dataset.create_customer()", use_standard_sql=True)
)
pipeLine.run().wait_until_fi
浏览 1提问于2021-11-06得票数 0
我正在尝试编写一个管道来读取来自pubsub的流，并使用google cloud dataflow和apache stream将其写入bigquery。我有这样的代码：
import apache_beam as beam
from apache_beam.transforms.window import FixedWindows
topic = 'projects/???/topics/???'
table = '???.???'
gcs_path = "gs://???"
with beam.Pipeline(runner="D
浏览 1提问于2017-11-29得票数 0
1回答
如何在中创建任务之间的依赖关系
、、、、
我是apache的新手，并探索apache数据流的python版本。我希望按一定顺序执行数据流任务，但它以并行模式执行所有任务。如何在中创建任务依赖关系？
示例代码：(在下面的代码中，sample.json文件包含5行)
import apache_beam as beam
import logging
from apache_beam.options.pipeline_options import PipelineOptions
class Sample(beam.PTransform):
def __init__(self, index):
self.index
浏览 0提问于2018-03-17得票数 0
我已经能够读入以下数据，将客户事务表示为带有光束的csv (Python SDK)。
timestamp,customer_id,amount
2018-02-08 12:04:36.899422,1,45.92615814813004
2019-04-05 07:40:17.873746,1,47.360044568200514
2019-07-27 04:37:48.060949,1,23.325754816230106
2017-05-18 15:46:41.654809,2,25.47369262400646
2018-08-08 03:59:05.791552,2,34.859367
浏览 4提问于2020-06-30得票数 0
我们运行了一个Google批处理作业，它将数据记录写入一个pubsub管道，并有一个单独的流作业，它使用“pubsub to elasticsearch”流模板从PubSub管道中提取数据，并将更新写入我们的Elasticsearch索引。但是，我们不得不终止写入Elastic的流作业，因为在1次尝试之后，我们遇到了多个“向ES写入错误”。再也不允许“错误”了。
从PubSub读取并写入Elasticsearch的任务是写入不是流索引的索引。它实际上需要在索引中更新相同的文档，可能会多次更新。
这项工作需要写入流索引吗？如果不是，我们如何配置数据流作业，使其实际上放慢速度，而不是压倒弹性？
我
浏览 16提问于2022-09-15得票数 0
当使用数据流运行beam模型时，我得到了以下错误。
java.lang.IllegalArgumentException: Class interface org.apache.beam.sdk.options.PipelineOptions missing a property named 'output'.
at org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1488)
at org.apache.beam.sdk.options
浏览 1提问于2019-09-17得票数 0
我注意到，RuntimeValueProvider在批处理模式示例中工作，但在带有流引擎的流模式下，运行时参数不会传输到作业，并且会引发错误
apache_beam.error.RuntimeValueProviderError: RuntimeValueProvider(option: templated_int, type: int, default_value: None).get() not called from a runtime context
想象一下流式模式下示例中的简单基本设置
import apache_beam as beam
from apache_beam.opti
浏览 18提问于2021-02-28得票数 0
2回答
当我从DirectRunner切换到Dataflowrunner：Pardo显然不起作用时，我遇到了问题。当我将runner设置为Dataflowrunner时，def process(self，query)从未运行。我可以看到这个工作在GCP上工作，但是，我的方法InsertPostgresql不使用Dataflowrunner。
根据日志错误，很明显，ParDo没有识别出“psycopg2 2”。
process NameError:在处理上述异常时未定义名称“psycopg2”
我想知道为什么？
def run_pipeline():
os.enviro
浏览 4提问于2020-11-11得票数 1
回答已采纳
2回答
我正试图在本地机器上运行一个python apache光束脚本来做一些模拟。我已经在我的选项中添加了'DirectRunner‘。但是p.run()给我一个错误"TypeError: Receiver()不接受参数“ 你知道为什么会发生这种情况吗？我使用Spyder作为IDE。 编辑:以下是一个代码示例，它采用以下形式的消息列表： { "Val_1": 1, "Val_2": 56, "date": "2019-04-01T15:00:04.340778" } 拆分它并将其放在 (1, 56, 2019-04-0
浏览 23提问于2019-04-02得票数 1
我正在尝试使用下面的代码读取Bigquery中的一些数据和文件系统中的一些数据。
( beam.io.Read(beam.io.BigQuerySource(query=apn_query，()( beam.combiners.ToList() beam.combiners.ToList() preprocess_rows =p= beam.io.ReadFromText(file_path，coder=UnicodeCoder())
但是，当我运行这个管道时，我会出现以下错误
文件"/etl/dataflow/etlTXLPreprocessor.py"，第125行，在ru
浏览 4提问于2020-03-01得票数 1
我的管道通过GCS通过Pub\Sub方式读取数据，然后将数据汇到redis。一开始，它在Dataflow中运行得很好。但是，在两天后运行后，在我的管道中发现了以下异常。
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.Snappy
org.xerial.snappy.SnappyOutputStream.<init>(SnappyOutputStream.java:97)
org.xerial.snappy.SnappyOutputStream.
浏览 1提问于2019-09-02得票数 0
回答已采纳
这个问题在周末开始出现。出于某种原因，这感觉像是DataFlow的问题。
以前，我能够很好地执行脚本和编写TF记录。但是，现在，我无法初始化计算图来处理数据。
回溯是：
Traceback (most recent call last):
File "my_script.py", line 1492, in <module>
MyBeamClass()
File "my_script.py", line 402, in __init__
self.run()
File "my_script.py", lin
浏览 4提问于2019-03-19得票数 1
我们希望使用函数参数add_value_provider_argument创建自定义数据流模板
如果不输入add_value_provider_argument ()中定义的变量，则无法启动以下命令
class UserOptions(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument(
'--input_topic',
help=
浏览 4提问于2019-11-13得票数 1
回答已采纳
1回答
我有一个非常小的python数据流包，包的结构如下所示
.
├── __pycache__
├── pubsubtobigq.py
├── requirements.txt
└── venv
requirements.txt的含量是
protobuf==3.11.2
protobuf3-to-dict==0.1.5
我用这个代码运行了我的对线
python -m pubsubtobigq \
--input_topic "projects/project_name/topics/topic_name" \
--job_name "job_name" \
浏览 2提问于2020-01-30得票数 0
我有一个简单的数据流管道，并试图从云shell执行，
代码：
from __future__ import print_function
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
with beam.Pipeline(options=PipelineOptions()) as p:
lines = p | 'Read' >> beam.io.ReadFromText('test.csv')
li
浏览 0提问于2019-03-30得票数 0
回答已采纳
当在本地运行我的Beam管道时，它完全按预期工作，但是当尝试在DataflowRunner上运行它时，我突然得到下面的错误。老实说，我甚至不知道从哪里开始评估它，因为DataflowRunner似乎是一个黑匣子。
Jan 14, 2019 11:26:51 AM org.apache.beam.runners.dataflow.DataflowRunner fromOptions
INFO: PipelineOptions.filesToStage was not specified. Defaulting to files from the classpath: will stage 165
浏览 0提问于2019-01-14得票数 1
我有一个数据流，它向API发出请求以检索一些数据。最近API中的密码有了更新，数据流突然开始失败。我使用的是java 1.8和beam SDK 2.19.0。同样的代码在本地运行时也可以正常运行。我尝试升级到java 11和same SDK 2.24.0，以防我使用的版本不支持新的密码，但我得到了相同的结果，它在本地运行，但我在数据流中得到了相同的错误。这是我用来向API发出请求的代码：
URL url = new URL(urlString);
HttpsURLConnection con = null;
outer: for (int ret
浏览 0提问于2020-11-05得票数 0
我正在尝试使用python中的Apache将BigQuery表记录作为JSON文件写入GCS桶中。
我有一张像这样的BigQuery表-- my_project.my_dataset.my_table
我希望将表记录/条目写入GCS桶位置中的JSON文件中- "gs://my_core_ bucket /data/my_data.json“
预期的JSON格式：
[
{"id":"1","values":{"name":"abc","address":"Mumbai
浏览 12提问于2021-07-12得票数 1
回答已采纳
1回答
当我通过以下步骤使用数据流时：- Read from bigquery - Convert table row to json string - Insert to elasticsearch (7.5.2)它看起来可以很好地处理大约100k的记录，但在实际(8m记录~65 im )的数据流中，300k插入后会抛出异常。
Error message from worker: java.lang.RuntimeException: unexpected org.apache.beam.runners.dataflow.worker.util.common.worker.CachingShuffl
浏览 21提问于2020-02-17得票数 0
回答已采纳
下面是我的python代码，它运行得非常好。
from __future__ import absolute_import
import apache_beam as beam
import argparse
import pickle
import logging
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.internal.clients import bigquery
from datetime import date
today = date.today
浏览 4提问于2021-02-03得票数 0
当我使用已定义的模板时，我总是遇到这个问题。我不确定问题出在哪里。 关于sdk：Apache Beam SDK for Java 2.10.0 Processing stuck in step WriteSuccessfulRecords/StreamingInserts/StreamingWriteTables/StreamingWrite for at least 05m00s without outputting or completing in state finish
at sun.misc.Unsafe.park(Native Method)
at java.util.c
浏览 11提问于2019-03-16得票数 0
我正在运行DataFlow配置
test_dataflow= BeamRunPythonPipelineOperator(
task_id="xxxx",
runner="DataflowRunner",
py_file=xxxxx,
pipeline_options = dataflow_options,
py_requirements=['apache-beam[gcp]==2.39.0'],
py_interpreter='python3',
dataflow_co
浏览 5提问于2022-07-08得票数 1
从服务器上，我能够连接并从配置了SSL的远程kafka服务器主题中获取数据。 在GCP中，如何使用Google Dataflow管道通过SSL信任库、密钥库证书位置和Google服务帐户json连接到远程kafka服务器？ 我使用Eclipse plugin for dataflow runner选项。 如果我指向GCS上的证书，当证书指向Google存储桶时，它会抛出错误。 Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: org.apache.kafka.
浏览 13提问于2019-01-24得票数 1
回答已采纳
我有一个在数据流中运行的。它是从PubSub读物和写到GCS。有时，在插入文件时，我会收到类似“_start_upload中的错误”之类的错误.“，它来自：
File "/usr/local/lib/python3.8/site-packages/apitools/base/py/base_api.py", line 603, in __ProcessHttpResponse raise exceptions.HttpError.FromResponse( RuntimeError: apitools.base.py.exceptions.HttpError: HttpErro
浏览 2提问于2021-06-14得票数 0
回答已采纳
1回答
Apache束流管道和毒丸
、、、、
我正在使用GCP数据流运行Apache Beam管道，并从worker那里获得了以下错误：
Error message from worker: java.lang.RuntimeException: java.lang.IllegalStateException: Got poison pill or timeout but stream is not done
两分钟内就有了。
我正在使用管道将消息从PubSub写到BigQuery。在管道中，当将PubSub消息转换为TableRow时，我使用的是FailsafeElement<PubsubMessage, String>，并
浏览 8提问于2022-09-15得票数 0
当我尝试通过将文档写入mongodb时 | "Write User Doc to Mongo" >> beam.io.WriteToMongoDB(uri=MONGO_URI,
db="dbname",
coll="col_name"
浏览 30提问于2019-09-04得票数 0
回答已采纳
我正在使用pubsub中的“”特性，通过数据流将一个常规的JSON传递给bigquery。
但是，它工作了一秒钟，这意味着一些条目可以正确地传递到bigquery。但是现在我在数据流日志上发现了错误
java.lang.RuntimeException: java.io.IOException:插入失败：[{“错误”：{“debugInfo”：“”、“位置”：“_comments”、“消息”：“没有这样的字段”、“原因”：“无效”}，“索引”：0}] org.apache.beam.sdk.io.gcp.bigquery.StreamingWriteFn.flushRows(Streamin
浏览 1提问于2018-03-27得票数 1
回答已采纳
2回答
将输出参数作为参数传递时，数据流管道作业失败，缺少消息输出属性。
错误：
Exception in thread "main" java.lang.IllegalArgumentException: Class interface org.apache.beam.runners.dataflow.options.DataflowPipelineOptions missing a property named 'output'.
at org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjec
浏览 0提问于2018-05-10得票数 1
我会在DataFlow运行器上执行我的Wordcount项目，我使用beam文档中的命令，但我得到了这个错误： C:\BIGDATA_FORMATION\TP_WC_GCP\word-count-beam>mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://bucket_wc/tmp/ --project=WordCount --inputFile=gs://buc
浏览 16提问于2020-09-02得票数 0
1回答
我正在学习Apache中的窗口&触发概念，目的是：
读取无界源(PubSub)
每5秒将传入消息写入本地主机磁盘，每5秒固定窗口间隔。
问题:没有输出被写入本地主机磁盘(管道确实创建了一个波束团队文件夹，并在其中写入了一些文件，但是目标目标中没有output.csv，每5秒编写一次)。
运行apache-beam=2.9.0，Python 2.7.10
尝试了两种方法：DirectRunner，和DataFlowRunner (以GCS桶为目的地)
下面是代码(谢谢您的建议)：
p = beam.Pipeline(runner=None, options
浏览 0提问于2019-01-02得票数 2
1回答
经过一天的正常工作，从发布/订阅流式传输数据，扁平数据并将行写入BigQuery；数据流管道开始报告如下错误： Processing stuck in step s01 for at least 05m00s without outputting or completing in state process
at sun.misc.Unsafe.park(Native Method)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.
浏览 9提问于2019-04-22得票数 1
我们正在运行一个简单的数据流作业，它从一个扳手DB中读取数据，并只显示已更改的记录的Mod类型。当我们在Dataflow中部署作业时，它总是失败，指出以下错误，
来自工作人员的错误消息:非便携式数据流不支持的org.apache.beam.runners.dataflow.worker.SplittableProcessFnFactory$SplittableDoFnRunnerFactory.lambda$createRunner$2(SplittableProcessFnFactory.java:172)\n : BundleFinalizer。\n org.apache.beam.ru
浏览 6提问于2022-08-18得票数 1
回答已采纳
1回答
我为Java开发人员使用Eclipse 4.7.3创建了一个基本的Google标准应用程序，该应用程序带有Dataflow管道。当我部署应用程序并调用web服务时，我在暂存包时得到java.lang.RuntimeException: Error。我做错了什么？以下是重现此错误的步骤：
使用创建一个新的Google项目(例如图钉-数据流测试)
使用为项目启用数据流API
使用为项目创建存储桶(例如，图钉-数据流测试)
运行Eclipse
在Google平台菜单中，选择>
在“项目名称”框中，键入项目名称(例如，DataflowTest)
在Java版本框中，选
浏览 5提问于2018-04-06得票数 0
回答已采纳
我想使用Java Beam SDK从worker节点以编程方式终止Apache Beam作业。理想情况下，我希望这个解决方案独立于运行者，但即使是特定于DataflowRunner的解决方案也可以。 我不想使用关闭钩子，我要找的是beam API支持的东西。 我找到的最接近我想要的是org.apache.beam.runners.dataflow.util.MonitoringUtil::getGcloudCancelCommand。但是，这只会返回一个字符串，其中包含需要运行以取消作业的命令。它不是从JVM中取消作业。
浏览 12提问于2020-03-18得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
