--在apache beam 0.6中缺少workerCacheMB设置？
Apache Beam是一个用于大规模数据处理的开源框架，它提供了一种统一的编程模型，可以在不同的分布式数据处理引擎上运行，如Apache Flink、Apache Spark和Google Cloud Dataflow等。
在Apache Beam 0.6版本中，确实没有提供workerCacheMB设置。workerCacheMB是用于设置工作节点的缓存大小的参数。缓存可以用于存储中间结果和其他临时数据，以提高计算性能。然而，在0.6版本中，Apache Beam还没有引入这个参数。
尽管在0.6版本中缺少了workerCacheMB设置，但可以通过其他方式来优化和管理工作节点的缓存。例如，可以使用分布式缓存系统（如Redis或Memcached）来存储和管理中间结果和临时数据。另外，可以通过调整工作节点的内存分配和任务调度策略来优化缓存效果。
需要注意的是，Apache Beam在后续的版本中可能会引入workerCacheMB设置或类似的功能，以进一步改进和优化数据处理性能。因此，建议在使用较新版本的Apache Beam时，查阅官方文档或社区讨论，以了解最新的功能和配置选项。
腾讯云提供了一系列与大数据处理和云计算相关的产品和服务，如腾讯云数据计算服务、腾讯云数据仓库、腾讯云弹性MapReduce等。您可以访问腾讯云官方网站（https://cloud.tencent.com/）了解更多信息。
相关·内容
我在一个便携式的Flink转轮上运行一个横梁作业
通过以下命令执行运行程序：
docker run -net=主机apache/beam_Flink1.10_job_server:最新-flink-master=localhost:8081
然后试图通过传递以下参数来运行梁作业：
--runner=PortableRunner --jobEndpoint=http://192.168.63.100:8099
按照这个设置作业。
它在启动时出现以下错误：
Exception in thread "main" java.lang.IllegalArgumentExcepti
浏览 13提问于2022-06-06得票数 0
1回答
当我尝试使用Apache编写BigQuery时，我得到了：
panic: element type is struct { TaskId string; Connector string; ... },
want struct { TaskId string "json:\"task_id\" bigquery:\"task_id\""; Connector string "json:\"connector\" bigquery:\"connector\""; ... }`
这是我的密码：
pac
浏览 5提问于2021-07-20得票数 1
1回答
我正在尝试创建一个数据流管道，但是在我的powershell控制台中得到了这个错误
Traceback (most recent call last):
File "load_pole_emploi.py", line 107, in <module>
run()
File "load_pole_emploi.py", line 92, in run
gcs_bucket_name + file_pattern)
File "E:\Utilisateurs\SI084342\venv\gcp\lib\site-pack
浏览 3提问于2022-04-22得票数 0
回答已采纳
使用GroupByKey的Hit problems，我相信可以追溯到一个类型问题。我已经研究了一段时间，并跟踪了一些堆栈跟踪，但我不清楚为什么下面的内容是错误的
@beam.typehints.with_output_types(beam.typehints.Tuple[long, float])
class MultiMap(beam.DoFn):
def process(self, element):
items = element.split(',')
print items
r = (long(items[0]), float
浏览 0提问于2018-06-30得票数 1
我正在尝试使用Java BigqueryIO.read方法从bigquery中读取数据。但是得到了低于错误。
public POutput expand(PBegin pBegin) { final String queryOperation = "select query"; return pBegin .apply(BigQueryIO.readTableRows().fromQuery(queryOperation)); }
2020-0
浏览 0提问于2020-06-09得票数 0
2回答
我有一个原型Apache管道，在这里我尝试使用以下配置从RabbitMQ读取数据
p.apply("read_from_rabbit", RabbitMqIO.read()
.withUri(options.getRabbitMQUri())
.withQueue(options.getRabbitMQQueue())
)
.apply("extract_json_data", MapElements.via(new Rabbi
浏览 0提问于2019-05-02得票数 1
回答已采纳
我得到了一个由maven编写的beam程序jar，我想用flink本地程序运行它。当我像这样跑的时候，没关系：
mvn :java -Dexec.mainClass=GroupbyTest -Dexec.args=“-runner=FlinkRunner
-flinkMaster=localhost:6123
--filesToStage=target/beamTest-1.0-SNAPSHOT.jar“
但是当我使用flink run时，protobuf有问题：
./bin/flink运行/home/maqy/Documents/beam_samples/beamTest/targ
浏览 0提问于2018-07-24得票数 1
回答已采纳
我的波束管道中有一个BigQueryIO.Write阶段，它是通过调用.withJsonSchema(String)来构造的。
inputStream.apply(
"save-to-bigquery",
BigQueryIO.<Event>write()
.withJsonSchema(jsonSchema)
.to((ValueInSingleWindow<Event> input) ->
new TableDestination(
浏览 7提问于2017-10-19得票数 5
回答已采纳
我正在做一个Google-Dataflow java项目。我试图在管道的初始化过程中创建和更新BigQuery表，所以为了开始，我尝试以这种方式实例化BigQuery Java客户端： BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService(); 在运行时，它会给我带来以下异常： Exception in thread "main" java.lang.NoClassDefFoundError: com/google/api/gax/retrying/ExceptionRetryAlgorith
浏览 23提问于2020-04-06得票数 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
试图做一些概念上简单的事情，但却让我的头撞到了墙上。
我尝试在Python语言中创建一个流式数据流作业，它使用来自PubSub主题/订阅的JSON消息，对每个消息执行一些基本操作(在本例中，将温度从C转换为F)，然后将记录发布回不同的主题：
from __future__ import absolute_import
import logging
import argparse
import apache_beam as beam
import apache_beam.transforms.window as window
import json
'''Normal
浏览 3提问于2018-06-29得票数 1
我有一个基本的Beam管道，它从GCS读取，做一个Beam转换，并将结果写到BigQuery。
当我没有在SQL语句中进行任何聚合时，它工作得很好：
..
PCollection<Row> outputStream =
sqlRows.apply(
"sql_transform",
SqlTransform.query("select views from PCOLLECTION"));
outputStream.setCo
浏览 1提问于2018-09-13得票数 0
我正在运行Python2.7，并试图安装Apache-Beam。
我运行这个命令: python -m pip install apache-beam。下面的输出。
当我尝试运行引用该库的代码时，安装肯定失败了，因为我得到了这个错误：“没有名为apache_beam的模块”
任何帮助都是非常感谢的。
以下是安装的输出：
(base) C:\Users\ned06>python -m pip install apache-beam
Collecting apache-beam
Collecting pyvcf<0.7.0,>=0.6.8 (from apache-beam)
浏览 44提问于2018-09-02得票数 0
1回答
您好，我在尝试使用Eclipse运行Google Dataflow模板示例Pub/Sub时遇到了这个错误。
Exception in thread "main" java.lang.Error: Unresolved compilation problem:
AutoValue_JavascriptTextTransformer_FailsafeJavascriptUdf cannot be resolved to a type
at com.google.cloud.teleport.templates.common.JavascriptTextTransformer$Fa
浏览 3提问于2018-09-26得票数 1
我有一条阿帕奇光束管道。在其中一个DoFn步骤中，它执行https调用(想想REST )。在我的本地环境中，所有这些都可以在DirectRun中很好地工作。
这是我的本地环境，apache 2.16.0：
$ mvn -version
Apache Maven 3.6.1 (d66c9c0b3152b2e69ee9bac180bb8fcc8e6af555; 2019-04-04T12:00:29-07:00)
Maven home: /opt/apache-maven-3.6.1
Java version: 1.8.0_222, vendor: Private Build, runtime:
浏览 2提问于2019-12-01得票数 2
我一直在使用一个Beam管道作为向导，试图从S3中为我的管道加载文件。就像在示例中一样，我定义了我自己的PipelineOptions，它也扩展了，我正在尝试使用。配置它的代码是：
MyPipelineOptions options = PipelineOptionsFactory.fromArgs(args).as(MyPipelineOptions.class);
options.setAwsCredentialsProvider(new DefaultAWSCredentialsProviderChain());
options.setAwsRegion("us-east-1&
浏览 0提问于2019-07-15得票数 1
回答已采纳
2回答
我使用来自的代码
并将其改为：
def getFullTableName(pn,tn):
return "{0}:{1}".format(pn,tn)
...
(
pipeline | "Read Data From Input Topic" >> beam.io.ReadFromPubSub(topic=data_topic)
| "Get Table data from input row" >> beam.Map(lambda r : data_ingestion.getData(r
浏览 12提问于2022-01-05得票数 0
我正在运行一个示例管道，我的环境是这样的。 python "SaiStudy - Apache-Beam-Spark.py“--runner=PortableRunner --job_endpoint=192.168.99.102:8099 我的Spark在Docker Container上运行，我可以看到JobService正在以8099的速度运行。 我收到以下错误: grpc._channel._MultiThreadedRendezvous:已终止的RPC的状态：<_MultiThreadedRendezvous = StatusCode.UNAVAILABLE deta
浏览 16提问于2020-10-24得票数 0
我试图编写一个示例，展示定时器和状态如何在ApacheBeam中工作，并使用beam操场测试这段代码。我读了一个数字列表，将它们以divisibleByTwo和divisibleByThree的形式存储在KV中。然后我用这个KV运行一个statefulDoFn。对于可被2整除的数字，我已经将计时器设置为在x秒之后过期，对于可被3整除的数字，我设置为y秒。但是，测试没有在x和y秒之后打印它们。我在测试用例中遗漏了什么吗？
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor li
浏览 4提问于2022-07-20得票数 1
回答已采纳
2回答
你能帮我解决这个问题吗。难道不可能将字符串的PCollection转换成行的集合吗？
难道不可能将字符串数组的集合转换为波束行的集合吗？
对于束模式中的所有字段，我尝试了String数据类型，但这也给了我同样的错误。
我使用的是Java 11、Maven 3.8.5和ApacheBEAMJavaSDK2.41.0
我用Java1.8和Beam 2.40.0尝试了相同的代码，得到了相同的错误。
public class beamRowPractise {
public static void main(String[] args){
PipelineOptions o
浏览 9提问于2022-10-14得票数 0
回答已采纳
1回答
是否可以在Beam中应用嵌套查询？我试图在Beam中运行这样的查询，但得到了错误。
我正在运行的查询是：
PCollection<BeamRecord> Query_Output = Query.apply(
BeamSql.queryMulti("Select Orders.OrderID From Orders Where Orders.CustomerID IN (Select Customers.CustomerID From Customers WHERE Customers.CustomerID = 2)"));
我得到的错误是：
浏览 1提问于2018-01-19得票数 0
回答已采纳
目前，我正在尝试从管道将数据提交到防火墙。
但我一直在犯错误：
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py", line 609, in do_work
work_executor.execute()
File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py", line 167, i
浏览 0提问于2018-06-20得票数 0
回答已采纳
1回答
LE: TL;DR；如何在Python中创建无界数据源？有可能吗？
我正在构建一个流数据流，它将继续处理来自具有时间戳、id和读取值的传感器的浮点值，将值放在FixedWindows中2秒，然后输出一个聚合。
代码链接：
这是快速vis的管道：
files = [
"in.csv",
]
fields = (p | beam.Create(files).with_output_types(str)
| beam.ParDo(FileReader())
| "Pars
浏览 1提问于2020-03-12得票数 2
我正在尝试使用连接两个无界源。当我加入的时候，我会收到以下错误信息。
线程"main“java.lang.UnsupportedOperationException中的异常:当前只支持非全局窗口使用已知每个窗口生成一次输出的触发器加入无界PCollections，例如默认触发器，允许延迟为零。在这种情况下，可保证每个窗口连接所有输入元素一次。( WindowingStrategy{windowFn=org.apache.beam.sdk.transforms.windowing.SlidingWindows@1b87117，allowedLateness=PT0S，trigger=R
浏览 0提问于2019-07-08得票数 1
回答已采纳
光束作业的流水线给出了以下异常
java.lang.RuntimeException: java.lang.RuntimeException: Exception while fetching side input:
org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:184)
org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunn
浏览 7提问于2019-11-22得票数 0
我正在尝试构建一个TableRow对象，最终将其写入到BigQuery表中，但是如果在行中包含一个NullPointerException值，则会得到一个null。这是完整的堆栈跟踪：
Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.NullPointerException
at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFin
浏览 18提问于2018-07-28得票数 1
回答已采纳
运行env：
2.12-2.0.1apach-beam OS:Ubuntu20.04kafka版本：库版本：apache-beam==2.32.0
操作步骤：
shell 1：在代码下面运行
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.external.kafka import ReadFromKafka
pipeline_options = PipelineOptions(["--runner=Direc
浏览 5提问于2021-09-01得票数 1
当我远程运行我的数据管道时，引发了一个PicklingError :数据管道是使用Python的Beam编写的，我正在Google Cloud Dataflow上运行它。当我在本地运行它时，管道工作得很好。
下面的代码生成PicklingError:这应该会重现这个问题
import apache_beam as beam
from apache_beam.transforms import pvalue
from apache_beam.io.fileio import _CompressionType
from apache_beam.utils.options import Pipeli
浏览 0提问于2016-10-26得票数 4
1回答
我正在flink集群上运行一个流束作业，在那里我得到了以下异常。 Caused by: org.apache.beam.sdk.util.UserCodeException: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:34)
浏览 24提问于2020-09-01得票数 0
1回答
嗨，我想创建systemctl服务到tun flink本地服务器。
我可以直接从/bin/start-local.sh运行它。
但是当我使用这个服务定义时。
[Unit]
Description=Apache Flink
Documentation=http://flink.apache.org/
Requires=network.target remote-fs.target
After=network.target remote-fs.target
[Service]
Type=simple
User=beam1
Environment=JAVA_HOME=/usr/lib/jvm/ja
浏览 0提问于2017-10-23得票数 3
1回答
我试图在GCS Dataflow上上传/运行一个Apache项目，但我得到了
INFO:apache_beam.runners.dataflow.internal.apiclient:Defaulting to the temp_location as staging_location: gs://dataflow-staging-europe-west4-10a485e03dda20c80122afcef299fc02
INFO:apache_beam.runners.dataflow.internal.apiclient:Starting GCS upload to gs://datafl
浏览 5提问于2022-03-29得票数 0
到目前为止，我有两个流水线在没有问题的产品上运行(都是n1-standard-4)。然而，当我决定尝试自动缩放时，它给出了上述错误。我尝试过使用普通的自动缩放和流媒体引擎( n2-highmem1，n1-highmen1)，但都不起作用。
这是我的管道的结构。这是BigQuery的Pubsub。
java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.OutOfMemoryError: Java heap space
org.apache.beam.runners.dataflow
浏览 0提问于2019-09-18得票数 0
使用下面的窗口函数，
Window.<KV<String, Long>>into(FixedWindows.of(Duration.standardDays(1)))
.triggering(
AfterWatermark.pastEndOfWindow()
.withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.s
浏览 0提问于2020-05-07得票数 1
我正在创建一个演示管道，使用我的免费谷歌帐户将CSV文件加载到BigQuery中。这就是我要面对的。
当我从GCS文件中读取并记录数据时，这是完美的。下面是我的示例代码。
，这段代码运行良好，
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setProject("project12345");
options.setStagingLocation("gs://mybucket/staging");
option
浏览 0提问于2019-07-08得票数 3
我正在使用Apache代码构建一个gradle java项目(请参阅下面)，并在Eclipse上执行。
package com.xxxx.beam;
import java.io.IOException;
import org.apache.beam.runners.spark.SparkContextOptions;
import org.apache.beam.runners.spark.SparkPipelineResult;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineRunner;
浏览 1提问于2018-05-21得票数 2
回答已采纳
我已经在一个码头上安装了apache_beam Python和airflow。Python版本: 3.5
阿帕奇气流: 1.10.5
我正在尝试使用**DataflowPythonOperator**执行apache波束管道。当我从气流用户界面运行DAG时，我得到
导入错误:导入apache_beam作为光束。模块未找到
在相同的设置下，我尝试了**DataflowTemplateOperator**，它运行得非常好。
当我尝试使用Python2和airflow 1.10.3进行相同的对接设置时，两个月前，操作符没有返回任何错误，而是按预期工作。
在SSH进入码头后，当我在一个码头容器
浏览 1提问于2019-09-06得票数 1
我在google云平台(GCP)中运行我的Google数据流工作。当我在本地运行这个作业时，它运行得很好，但是在GCP上运行时，我得到了一个错误"java.lang.IllegalArgumentException:找不到方案gs的文件系统“。我可以访问那个google，我可以将我的jar文件上传到那个URI中，并且我可以看到一些临时文件用于我的本地工作。
我在GCP的工作证明：
2019年-08-08_21_47_27-162804342585245230(束流版本:2.12.0)
2019-08-09_16_41_15-11728697820819900062 (beam版本:2.
浏览 1提问于2019-08-10得票数 5
回答已采纳
5回答
主要目标是聚合两个Kafka主题，一个是压缩的慢速移动数据，另一个是每秒接收的快速移动数据。
我已经能够在KV (Long，String)这样的简单场景中使用如下内容来消费消息：
PCollection<KV<Long,String>> input = p.apply(KafkaIO.<Long,
String>read()
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
PCollection<String&
浏览 7提问于2019-02-19得票数 2
我正在尝试遵循来处理代码中的BigQuery错误。
PCollection convertedTableRows =
pipline
.apply("ReadFromKafka", buildReadToKafkaIO(options))
.apply("ConvertMessageToTableRow", new.TransformTableRow());
WriteResult writeResult = convertedTableRows.apply("Wr
浏览 10提问于2020-05-22得票数 0
我有一个数据流作业，它连接到cloudsql并持久化一些数据。我平均有大约75个活跃的连接(偶尔会有几个峰值到100多个连接)。因此，我想知道是否存在连接的最大数量。文档似乎并没有表明。()
背景故事和一些背景:我的一个作业出现错误，它似乎只是随机锁定并停止持久化数据：
Operation ongoing in step X for at least 305h20m00s without outputting or completing in state start
at sun.misc.Unsafe.park (Native Method)
at java.util.concurrent.
浏览 0提问于2021-05-05得票数 1
2回答
这个问题是的后续问题.我正在尝试使用apache从google扳手表中读取数据(然后进行一些数据处理)。我使用java编写了以下最小示例：
package com.google.cloud.dataflow.examples;
import java.io.IOException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sd
浏览 6提问于2017-10-11得票数 4
2回答
我使用的是ClouderaEnterprise6.1.0版本的，在用SparkRunner读取或编写HDFS上的任何文件时，使用apache 2.11SDKS来解决这个问题。但是，有了火花，它就开始运作了。
这个问题是在将Cloudera版本从5.14.0升级到6.1.0后出现的，在以前的版本中，它可以很好地处理下面的代码。
import java.io.File;
import java.io.IOException;
import java.sql.ResultSet;
import org.apache.beam.runners.spark.SparkContextOptions;
im
浏览 0提问于2019-03-13得票数 2
1回答
我试图在中运行一个Apache应用程序，它使用Apache作为运行时。管道使用连接器。我正在尝试，因为Kinesis不允许导出环境变量，导出环境变量似乎不是一种选择。
我试图使用下面的代码进行身份验证。
GoogleCredentials credential = GoogleCredentials
.fromStream(credentialJsonInputStream)
.createScoped("https://www.googleapis.com/auth/cloud-platform", "https
浏览 4提问于2021-08-13得票数 1
我将向kafka输入多个表的数据，beam将在获得数据后执行SQL，但现在有以下错误：
线程"main“中出现异常
org.apache.beam.sdk.values.PCollection.getSchema(PCollection.java:328) org.apache.beam.sdk.extensions.sql.impl.schema.BeamPCollectionTable.(BeamPCollectionTable.java:34) at org.apache.beam.sdk.extensions.sql.SqlTransform.toTableMap(SqlTra
浏览 10提问于2019-11-22得票数 0
回答已采纳
我正试图用Beam 2.20.0版本在Python3.7中构建一个Apache管道，该管道成功地部署在Dataflow上，但似乎没有做任何事情。在worker日志中，我可以看到以下重复报告的错误消息
同步pod ()错误，跳过:启动容器失败
我已经尝试了我所能做的一切，但是这个错误很顽固，我的管道看起来是这样的。
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options impo
浏览 1提问于2020-05-17得票数 1
请参阅有关如何运行基于Java的云数据流- 的链接。
使用以下命令创建模板
mvn archetype:generate \
-DarchetypeGroupId=org.apache.beam \
-DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
-DarchetypeVersion=2.16.0 \
-DgroupId=org.example \
-DartifactId=word-count-beam \
-Dversion="0
浏览 2提问于2020-04-21得票数 0
回答已采纳
1回答
我们的SDK版本是Apache Beam Python 3.7 SDK 2.25.0。 有一个从Pub/Sub读取数据、转换数据并将结果保存到GCS的管道。通常它在1-2周内工作良好。在那之后，它就变得糟糕了。 "Operation ongoing 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.par
浏览 19提问于2020-12-10得票数 2
1回答
我有一个非常简单的数据流工作，我想编写单元测试。遗憾的是，没有好的例子说明什么是最好的方法。
这是代码
import logging
from datetime import datetime
from re import sub
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from beam_nuggets.io im
浏览 5提问于2020-01-02得票数 2
2回答
我一直试图让我的管道运行一个典型的模板在Dataflow。
管道应该读取REST。然后，从API返回的答案应该写入bigquery表中。
当我将api限制为1结果时，它正在成功运行，但是，如果我将限制增加到1以上，它将作为错误返回。
非常感谢您的帮助--这是我的管道代码：
#!/usr/bin/env python3
import apache_beam as beam
#from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options i
浏览 1提问于2021-11-26得票数 0
2回答
为流构建此GCP数据流示例时出错
、、、、
我正在尝试构建将Pub/Sub流到BigQuery的以下示例：
守则是：
/*
* Copyright (C) 2018 Google Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy of
* the License at
*
* http://www.apache.org/licenses
浏览 4提问于2019-04-12得票数 2
回答已采纳
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频
腾讯云开发者
