从数据流sdk切换到beam sdk导致URN查找异常
从数据流 SDK 切换到 Beam SDK 导致 URN 查找异常是因为两个 SDK 在处理数据流时使用了不同的 URN（Uniform Resource Name）查找机制。
数据流 SDK 是一种用于处理实时数据流的开发工具包，它提供了一套用于构建和管理数据流处理应用程序的 API。在数据流 SDK 中，URN 是用于标识和查找数据流处理器（如转换器、窗口函数等）的唯一标识符。当切换到 Beam SDK 后，由于 Beam SDK 使用了不同的编程模型和数据流处理框架，URN 的查找机制也发生了变化，导致之前使用数据流 SDK 的 URN 查找异常。
Beam SDK 是一种用于批处理和流处理的开发工具包，它提供了一套统一的编程模型和 API，可以在不同的数据处理引擎上运行。在 Beam SDK 中，URN 的查找是通过注册表和命名空间来实现的。注册表是一个存储 URN 和对应处理器实现的映射关系的数据结构，而命名空间则用于区分不同的处理器类型。当切换到 Beam SDK 后，需要根据新的 URN 查找机制来更新代码，以适应 Beam SDK 的处理方式。
为了解决 URN 查找异常，可以按照以下步骤进行操作：
- 确认切换到 Beam SDK 的原因和目的，了解 Beam SDK 的特性和优势。
- 检查代码中使用的 URN 查找逻辑，查找与数据流 SDK 相关的部分。
- 根据 Beam SDK 的文档和示例代码，了解新的 URN 查找机制和使用方式。
- 更新代码中的 URN 查找逻辑，使用 Beam SDK 提供的注册表和命名空间进行查找。
- 运行和测试更新后的代码，确保 URN 查找异常问题已解决。
在腾讯云的产品中，与数据流处理相关的产品包括腾讯云流计算（Tencent Cloud StreamCompute）和腾讯云数据流服务（Tencent Cloud Data Stream Service）。这些产品提供了基于云原生架构的数据流处理能力，可以帮助开发者快速构建和部署实时数据处理应用程序。
腾讯云流计算是一种实时数据处理引擎，支持基于 SQL 和编程模型的数据流处理。它提供了丰富的数据处理算子和函数库，可以方便地进行数据转换、聚合、过滤等操作。腾讯云流计算还提供了可视化的开发工具和调试工具，帮助开发者快速开发和调试数据流处理应用程序。
腾讯云数据流服务是一种全托管的数据流处理服务，提供了可弹性扩展的计算资源和高可靠的数据处理能力。它支持基于 Apache Flink 的数据流处理引擎，可以处理大规模的实时数据流。腾讯云数据流服务还提供了丰富的数据源和数据接收器，可以与其他腾讯云产品（如腾讯云对象存储、腾讯云数据库等）进行集成，实现全链路的数据处理和分析。
更多关于腾讯云流计算和腾讯云数据流服务的详细信息和产品介绍，请参考以下链接：
相关·内容
我想使用Java Beam SDK从worker节点以编程方式终止Apache Beam作业。理想情况下，我希望这个解决方案独立于运行者，但即使是特定于DataflowRunner的解决方案也可以。 我不想使用关闭钩子，我要找的是beam API支持的东西。 我找到的最接近我想要的是org.apache.beam.runners.dataflow.util.MonitoringUtil::getGcloudCancelCommand。但是，这只会返回一个字符串，其中包含需要运行以取消作业的命令。它不是从JVM中取消作业。
浏览 12提问于2020-03-18得票数 0
回答已采纳
我尝试运行数据流管道，从本地机器(Windows)读取数据，并使用DirectPipelineRunner将数据写入谷歌云存储。作业失败，错误如下:指定FileNotFoundException(因此我认为数据流作业无法读取我的位置)。我从本地计算机运行该作业，以运行我创建的基于GCP的模板。我可以在GCP Dataflow仪表板中看到它，但失败并显示以下错误。请帮帮忙。我也尝试了我的本地机器的IP或主机名以及我的本地位置，但面对这个FileNotFoundException？
错误：
java.io.FileNotFoundException: No files matched spec:
浏览 4提问于2018-08-10得票数 1
我们从dataflow sdk (2.5.0)改为apache beam sdk (2.12.0)，代码没有变化。 我们会得到以下错误消息 Expected DoFn to be FunctionSpec with URN urn:beam:dofn:javasdk:0.1, but URN was 还需要做什么其他的设置吗？ 这条线路似乎失败了。 pipeline.apply(<some code here>).watchForNewFiles(10, Watch.Growth.never()))
浏览 14提问于2019-05-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
回答已采纳
我正在尝试创建一个google数据流模板，但我似乎无法找到一种不产生以下异常的方法：
WARNING: Size estimation of the source failed: RuntimeValueProvider{propertyName=inputFile, default=null}
java.lang.IllegalStateException: Value only available at runtime, but accessed from a non-runtime context: RuntimeValueProvider{propertyName=inputFile,
浏览 0提问于2018-05-15得票数 5
当在本地运行我的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
当我使用已定义的模板时，我总是遇到这个问题。我不确定问题出在哪里。 关于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
我想在Google上读取来自MariaDB的数据，并通过DataFlow将数据写入BigQuery，但在DataFlowRunner上运行DataFlow程序时，我总是会遇到以下异常。
address=(host=xxx.xxx.xxx.xxx)(port=3306)(type=master)：java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.sql.SQLException:无法创建PoolableConnectionFactory (无法连接到org.apache.beam.sdk.u
浏览 1提问于2018-03-14得票数 0
回答已采纳
我正在尝试在Google Cloud Dataflow上运行作业，但无法使部署工作。使用DirectRunner可以很好地运行作业，但是一旦我切换到dataflow-runner，我就会得到以下异常：
[WARNING]
java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccesso
浏览 0提问于2017-04-05得票数 1
我正在使用apache_beam==2.24.0编写一个用python2.7编写的数据流管道。该管道的工作是使用beam的ReadFromPubSub批量使用订阅中的pubsub消息，对消息进行一些处理，然后将结果数据持久化到两个不同的bigquery表中。有很多我正在消费的数据。谷歌-云-公开版是1.7.0。在运行管道之后，一切正常，但几个小时后，我开始得到异常：
已取消的org.apache.beam.vendor.grpc.v1p13p1.io.grpc.StatusRuntimeException：：已取消的电话
在gcp数据流控制台上，日志显示此错误，但作业本身似乎运行良好。它使用
浏览 1提问于2021-03-31得票数 0
我正在运行一个数据流作业来从pubsub获取数据，并将其传递给大型查询。但在此过程中，我在运行1小时后收到以下错误。 INFO:oauth2client.transport:Refreshing due to a 401 (attempt 1/2)
INFO:oauth2client.client:Refreshing access_token
INFO:root:2018-12-19T09:13:09.510Z: JOB_MESSAGE_ERROR:
java.util.concurrent.ExecutionException: org.apache.beam.vendor.grpc.
浏览 89提问于2018-12-19得票数 1
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
我们运行了一个Google批处理作业，它将数据记录写入一个pubsub管道，并有一个单独的流作业，它使用“pubsub to elasticsearch”流模板从PubSub管道中提取数据，并将更新写入我们的Elasticsearch索引。但是，我们不得不终止写入Elastic的流作业，因为在1次尝试之后，我们遇到了多个“向ES写入错误”。再也不允许“错误”了。
从PubSub读取并写入Elasticsearch的任务是写入不是流索引的索引。它实际上需要在索引中更新相同的文档，可能会多次更新。
这项工作需要写入流索引吗？如果不是，我们如何配置数据流作业，使其实际上放慢速度，而不是压倒弹性？
我
浏览 16提问于2022-09-15得票数 0
我的管道通过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运行器上执行我的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
我有一个相当简单的数据流作业，它从BigQuery读取并写入到一个扳手实例。所有读取都已完成，但由于某些原因，在开始向扳手写入时，它始终出现EOFException错误，并且它位于类MutationGroupEncoder中。大约有580M行和1.36 it的数据已经被分区，但是在写入大约3500个之后，它失败了。
下面的异常跟踪-堆栈跟踪中没有涉及我们的代码，所有的库代码都在编组输出数据。我还没能找到其他人报告类似的bug。我们使用的是Google Cloud Apache Beam SDK的2.5.0版本。
java.lang.RuntimeException: org.apac
浏览 3提问于2018-07-26得票数 2
我创建了一个google流作业，从PubSub读取并插入到BigQuery中。我使用JSON将STREAMING_INSERT数据插入到BiqQuery表中。我正面临插入问题，说明请求的大小超过了10 Mb的允许限制。数据流错误如下所示。根据我的理解，每个记录的大小是1-2Mb，根据我的理解，数据流作业将流数据插入到微批处理中，这导致了这一错误。
你能不能给我一些解决办法。
来自worker的错误消息: java.lang.RuntimeException:我们观察到一行大小为24625273字节。BigQuery支持高达10 is的请求大小，这一行太大了。您可以更改重试策略，以取消阻塞该管
浏览 6提问于2022-10-13得票数 0
回答已采纳
1回答
我为Java开发人员使用Eclipse 4.7.3创建了一个基本的Google标准应用程序，该应用程序带有Dataflow管道。当我部署应用程序并调用web服务时，我在暂存包时得到java.lang.RuntimeException: Error。我做错了什么？以下是重现此错误的步骤：
使用创建一个新的Google项目(例如图钉-数据流测试)
使用为项目启用数据流API
使用为项目创建存储桶(例如，图钉-数据流测试)
运行Eclipse
在Google平台菜单中，选择>
在“项目名称”框中，键入项目名称(例如，DataflowTest)
在Java版本框中，选
浏览 5提问于2018-04-06得票数 0
回答已采纳
我想从WordCount上运行 java示例，但是不知怎么地，我得到了一个错误，ClasspathScanningResourcesDetector没有找到任何要处理的文件。我运行的示例与网站上描述的完全相同：
mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
-Dexec.args="--runner=DataflowRunner --project=<your-gcp-project> \
--gcpTempLocatio
浏览 3提问于2020-03-08得票数 2
回答已采纳
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
从服务器上，我能够连接并从配置了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
回答已采纳
我试图在Eclipse中(使用本地运行程序)在本地运行，这是一个使用Apache的Java数据流。
当我添加pubsub作为数据源时，就会出现这个问题。
(如果我删除该行并尝试使用一个简单的管道，我可以毫无问题地执行它)
管道代码：
public class StarterPipeline {
public static void main(String[] args) {
Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
run(optio
浏览 1提问于2019-12-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
回答已采纳
1回答
我是数据流的新手，并且尝试在大查询中动态获取表的模式。此外，我还需要在BigQueryIO.write.to()中使用动态目标类来动态获取目标表的名称。如果在执行管道之前为目标表提供了模式，那么它就会起作用。但是为了动态地获取模式，我使用了BigQuery片段，它接受datasetId和tableId作为输入，并返回给定表的模式。当尝试运行带有代码段的管道时，它会给出下面提到的错误。
任何帮助都是非常感谢的。提前谢谢。
Exception in thread "main" java.lang.NoSuchMethodError: com.google.api.client.go
浏览 2提问于2018-07-06得票数 0
我正试图用我的自定义对象创建一个虚拟PCollection，如下所示：
PCollection<MyClass> pipelineProcessingResults = pipeline.apply(Create.of(new MyClass(.., ..)));
MyClass类如下：
@DefaultSchema(JavaBeanSchema.class)
public class MyClass {
AnotherComplexClass _obj;
Urn _urn
}
我得到了以下例外：
java.lang.StackOverflowError
at j
浏览 5提问于2022-07-12得票数 0
回答已采纳
1回答
我试图在google数据流上运行一个有状态聚合DoFn，它在其能力矩阵中引用有状态DoFns，但是我得到了以下错误：
异常:请求执行有状态DoFn，但没有可用的用户状态上下文。这可能意味着当前运行程序不支持有状态DoFns的执行。
前面的错误发生在这里：
@with_input_types(Dict[K, V])
@with_output_types(Dict[K, V])
class StatefulCombineDoFn(beam.DoFn):
BUFFER = BagStateSpec(
'buffer',
Pickle
浏览 0提问于2019-06-01得票数 2
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
我想使用Dataflow将数据从发布/订阅移动到GCS。因此，基本上我希望Dataflow在固定的时间内(例如15分钟)积累一些消息，然后在该时间段过去后将这些数据作为文本文件写入GCS。 我的最终目标是创建一个自定义的管道，所以“发布/订阅到云存储”模板对我来说是不够的，而且我对Java一无所知，这让我开始在Python中进行调整。 这是我目前所得到的(Apache Beam Python SDK 2.10.0)： import apache_beam as beam
TOPIC_PATH="projects/<my-project>/topics/<my-to
浏览 6提问于2019-02-18得票数 7
回答已采纳
当使用数据流运行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
1回答
我正在编写apache管道，用于处理python中的大数据文件。该管道的参数由argparse和PipelineOptions管理。我在_add_argparse_args()中创建了一个详细的参数(用于处理日志级别)
parser.add_argument('-v', '--verbosity', action='count', help='Increase output verbosity', default=0)
此参数允许为日志信息指定-v，为日志调试指定-vv，为log指定-vvv。当未指定参数时，只显示警告和错误日志。
浏览 5提问于2022-04-04得票数 0
在使用ValueProvider执行run命令时，我尝试获取服务器ID作为参数 对于选项界面中的值提供程序： ValueProvider<String> getKafkaServer();
void setKafkaServer(ValueProvider<String> value); withBootstrapServers抛出错误“不兼容的类型:无法将org.apache.beam.sdk.options.ValueProvider转换为java.lang.String” PCollection<String> kafkaMessages =
浏览 47提问于2021-10-01得票数 0
我试图在python中构建一个流束管道，它应该从kafka捕获消息，然后执行从其他源获取数据和聚合的进一步阶段。到目前为止，我逐步建立的过程是：
在本地主机上运行Kafka实例的：9092
./bin/kafka-server-start.sh ./config/server.properties
运行波束-使用码头的flink作业服务器
docker run --net=host apache/beam_flink1.10_job_server:latest
运行梁-卡夫卡管道
import apache_beam as beam
from apache_beam.io.external.k
浏览 3提问于2020-07-14得票数 3
2回答
我有一个成功的管道，使用在计算实例上运行的云dataflow将数据从pub/sub传输到bigquery，而不是在实际的dataflow运行器上运行。
今天我已经更新了BQ表模式，似乎没有发生新的插入。我可以查看机器上的日志，一切正常- dataflow没有报告任何错误。
有没有办法从bigquery访问流日志以检查错误。
编辑:总结我的问题是，我是否能够从apache beam SDK或bigquery获得更详细的日志记录，以查看这些数据的结束位置。
我在stackdriver中查看了一下，但这似乎没有为流日志创建条目。
浏览 22提问于2019-11-22得票数 2
回答已采纳
1回答
我正在运行一个数据流作业(jobid: 2018-03-13_13_21_09-13427670439683219454)。
作业在1小时后停止运行，并有以下错误消息：
(f38a2b0cb8c28493): Workflow failed. Causes: [...]
(6bf57c531051aa32): A work item was attempted 4 times without success.
Each time the worker eventually lost contact with the service.
The work item was attempted
浏览 1提问于2018-03-13得票数 1
回答已采纳
2回答
因此，我在Apache中实现了这个作业，以便最终在Dataflow中运行它。所以我用Direct进行了测试，但是当我将它更改为Dataflow Runner时，它崩溃了：
java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0 (Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke (NativeMethodAccessorImpl.java:62)
at sun.refle
浏览 1提问于2018-01-30得票数 0
回答已采纳
我正在构建一个Apache管道，作为一个无限制的源从Kafka读取。
我能够使用直接跑步器在本地运行。
但是，当在云中使用运行时，附加的异常堆栈跟踪将导致管道失败。
似乎最终是抛出了。我不太清楚该如何解决这个问题。
java.io.IOException: Failed to start reading from source: org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@33b5ff70
com.google.cloud.dataflow.worker.WorkerCustomSources$UnboundedReade
浏览 0提问于2018-11-05得票数 5
回答已采纳
我有一个简单的数据流管道，并试图从云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
回答已采纳
我正在使用以下命令在云存储中创建和暂存gcp数据流模板：
mvn -X compile exec:java -Dexec.mainClass=main.java.TemplatePipeline -Dexec.args="--runner=DataflowRunner --project=hv-hcap-development --stagingLocation=gs://my-bucket/staging --templateLocation=gs://my-bucket/templates/template1"
构建失败，出现以下错误
[INFO] BUILD FAILUR
浏览 3提问于2020-07-18得票数 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
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
3回答
我有一个数据流作业，在这个用户代理中运行OK (我们只配置Apache版本，这样JDK环境就会自动配置)
Apache_Beam_SDK_for_Java/2.19.0(JDK_11_environment)
升级工作流后，我们使用Beam 2.27声明，作业由以下用户代理启动：
Apache_Beam_SDK_for_Java/2.27.0(JDK_11_environment)
它开始因错误而失败：
工作流失败。原因:找不到订阅'xxx‘。请提供现有订阅。
发布子订阅是存在的，它与我们以前使用的相同，甚至作业配置属性中的链接也运行良好，并导航到订阅。
以前，我们已经成功地
浏览 0提问于2021-04-27得票数 2
1回答
我正在尝试编写一个简单的Apache管道(它将在Dataflow运行程序上运行)来执行以下操作：
从subscription.For (每条消息)读取包含GCS上文件路径的PubSub消息，读取与消息相关的文件中包含的数据(文件可以是各种格式(csv、jsonl、json、xml、.)。对每条记录做一些处理。在GCS.上写结果。
我用的是10秒的固定窗口。由于传入的文件已经分块了(最大大小为10 max)，所以我决定不使用可拆分的do函数来读取文件，以避免增加无用的复杂性(特别是对于不能以块分割的文件)。
下面是一个简化的代码示例，它给出了与完整代码完全相同的问题：
package skytv
浏览 7提问于2021-01-25得票数 1
我们正在创建一个数据流管道，我们将从postgres读取数据并将其写入一个拼花文件。我们使用org.apache.beam.sdk.io.jdbc读取文件，使用org.apache.beam.sdk.io.parquet包写入文件。ParquetIO.Sink允许您将PCollection of GenericRecord写入Parquet文件(从这里开始，)。
到目前为止，这是我的代码：
Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());
Schema
浏览 1提问于2019-06-26得票数 2
回答已采纳
我有一个用java编写的项目，其中包含多个波束管道，我将其编译为jar文件，以便在服务器上执行。目前一切正常，我只是从GCP资源中读取，但我只是添加了一个写入S3的管道。S3部件独立工作，但现在当我尝试运行其他只使用GCP的管道时，它抛出一个异常，因为我没有提供S3选项(即使我不需要它们)-下面的错误消息。当我只使用GCP资源(或者我做错了什么)时，我需要指定一个AWS区域，这似乎有点不对劲。有没有办法在初始化时只注册用于特定管道的文件系统，而不是全面注册所有文件系统？ INFO: The AWS S3 Beam extension was included in this build, b
浏览 45提问于2019-01-05得票数 1
回答已采纳
2回答
我创建一个maven项目来执行管道。如果我运行主类，管道就能很好地工作。如果我创建一个胖jar并执行它，我会有两个不同的错误，一个在Windows下执行，另一个在Linux下执行。
在Windows下：
Exception in thread "main" java.lang.RuntimeException: Error while staging packages
at org.apache.beam.runners.dataflow.util.PackageUtil.stageClasspathElements(PackageUtil.java:364)
浏览 0提问于2018-02-08得票数 1
1回答
在阅读了之后，我仍然有一些关于Dataflow/Apache如何分配工作负载的问题。下面的代码可以演示我遇到的问题：
package debug;
import java.io.IOException;
import org.apache.beam.runners.dataflow.DataflowRunner;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.runners.dataflow.options.DataflowPipelineWor
浏览 0提问于2018-08-02得票数 4
以下代码：
def get_pipeline(workers):
pipeline_options = PipelineOptions(['--direct_num_workers', str(workers)])
return beam.Pipeline(options=pipeline_options,
runner=fn_api_runner.FnApiRunner(
default_environment=beam_runner_api_pb2.E
浏览 19提问于2019-11-05得票数 3
我有一个在数据流中运行的。它是从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回答
在我的数据流(beam)工作流中，我使用来自Python的datetime包(在gcp上使用jupyter notebook )。当我用pip安装缺少的包时，一切都很好。
现在我想在gcp上运行我的转换为数据流作业。
我总是得到错误的"NAME ERROR“，因为datetime是未知的(当我使用Apache beam SDK for Python时得到这个错误)，有人能给我一个提示如何将Python包添加到数据流作业吗？
我使用的是setup.py文件，但它不起作用。
浏览 1提问于2020-11-16得票数 0
1回答
docs提到了以下选项：direct_num_workers和direct_running_mode，以及设置streaming选项。 所有这些都在DirectOptions class中缺失了 此外，当尝试从args设置这些值时，会抛出以下异常： java.lang.IllegalArgumentException: Class interface org.apache.beam.runners.direct.DirectOptions missing a property named 'direct_running_mode'.
at org.apache
浏览 9提问于2020-10-04得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
