在Apache Beam上传递PCollection作为侧输入的KeyError
Apache Beam是一个用于大规模数据处理的开源框架，它提供了一种统一的编程模型，可以在不同的分布式处理引擎上运行，如Apache Flink、Apache Spark和Google Cloud Dataflow等。在Apache Beam中，PCollection是数据处理的核心概念，代表了一个数据集合。
在Apache Beam中，PCollection可以通过侧输入（Side Input）的方式传递给数据处理函数。侧输入是一种额外的输入，用于在数据处理过程中提供额外的信息。通常情况下，侧输入是一个键值对（Key-Value）的集合，其中键用于与主输入进行关联，值则是与键相关的附加数据。
在给定的问答内容中，提到了一个KeyError。KeyError是Python编程语言中的一个异常类型，表示在字典或集合中查找指定键时未找到该键。然而，在Apache Beam中并没有直接与KeyError相关的概念或异常。
总结起来，Apache Beam是一个用于大规模数据处理的开源框架，PCollection是其核心概念之一，用于表示数据集合。侧输入是一种传递给数据处理函数的额外输入，用于提供附加信息。KeyError是Python编程语言中的异常类型，与Apache Beam并无直接关联。
腾讯云提供了一系列与大数据处理相关的产品和服务，如腾讯云数据工场、腾讯云数据湖、腾讯云数据仓库等，可以帮助用户在云上进行大规模数据处理和分析。具体产品介绍和链接地址可以参考腾讯云官方网站的相关页面。
相关·内容
我将side_input PCollection作为侧输入传递给ParDo转换，但是为了同样的目的获得了KeyError import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import relational_db
from processors.appendcol import AppendCol
from side_inputs.config import sideinput_bq_config
from sourc
浏览 12提问于2020-09-26得票数 1
回答已采纳
我有两个PCollections:一个从Pub/Sub获取信息，另一个从CSV文件中提取数据。在每个管道中进行了一些不同的转换之后，我想将两者合并在一个公共密钥上，它们都是共享的，"STREET“。我将第二个PCollection作为侧输入。但是，我在尝试运行时出错了。
我试图利用CoGroupByKey，但是我一直收到关于Pcollections中数据类型不同的错误。我尝试重构输出，并通过PCollection设置__setattr__属性，以强制类型相等，但它报告了“混合值”。经过进一步的研究，似乎更好地利用了侧输入，特别是当元素之间的数据大小存在差异时。即使有侧输入，我仍然无法通
浏览 0提问于2019-04-02得票数 3
回答已采纳
使用Apache的Go SDK，我尝试使用一个侧输入来创建一个PCollection视图。
但我发现了一个奇怪的错误：
Failed to execute job: on ctx= making side input 0:
singleton side input Singleton for int ill-defined
exit status 1
在这里，我使用的代码：
// A PCollection of key/value pairs
pairedWithOne := beam.ParDo(s, func(r models.Review) (string, int) {
浏览 0提问于2019-09-09得票数 2
回答已采纳
我有一个用例，在这个用例中，我需要将一个file_path (作为一个新元素)附加到另一个pcollection中，这是一个元素字典。
因此，我决定将pcollection作为下一个转换的输入，将file_path (由pardo函数生成)作为侧输入。
生成的file_path是一个pCollection，所以我将它用作：
beam.pvalue.AsSingelton(file_path)
但奇怪的是，它没有执行这个语句，因此没有输出，甚至没有任何错误。但是，当我使用以下方法创建pcollection时：
beam.Create('some_path_value') file
浏览 3提问于2019-08-29得票数 1
我正在尝试用Apache创建一个管道，第一步是通过取输入PCollection的平均值，然后用一个Map从平均值中减去每个元素，从而将输入时间序列中心设为0。但是，运行以下脚本将导致以下错误：
'_InvalidUnpickledPCollection' object has no attribute 'windowing'
import apache_beam as beam
import numpy as np
from apache_beam.testing.test_pipeline import TestPipeline
raw_input = n
浏览 6提问于2022-08-01得票数 0
回答已采纳
我使用Apache (SDK2.40.0)和GCP数据流流道和流管道。我需要使用配置来处理可以随时更改的数据。因此，我每2分钟加载一次(可接受的延迟)，作为一个侧输入，如下所示：
configs = (
p
| PeriodicImpulse(fire_interval=120, apply_windowing=False)
| "Global Window" >> beam.WindowInto(
window.GlobalWindows(),
trig
浏览 6提问于2022-10-12得票数 0
1回答
与侧输入连接
、、、
我可以在Apache中加入一个PCollection和一个PCollectionView (通过侧输入)吗？
在中，在查询中实现联接的方法是创建一个只接受PCollections的PCollectionTuple。
// Create a PCollectionTuple containing both PCollections.
// TupleTags IDs will be used as table names in the SQL query
PCollectionTuple namesAndFoods = PCollectionTuple
.of(new TupleTag&
浏览 5提问于2022-03-29得票数 0
使用“数据流作业”从bigQuery表读取数据时，要尽量避免集合中的重复。对于这一点，使用beam.sdk.transforms.Distinct来读取带有distinct的记录。但却在错误之下
java.lang.IllegalStateException: the keyCoder of a GroupByKey must be deterministic
at org.apache.beam.sdk.transforms.GroupByKey.expand(GroupByKey.java:193)
at org.apache.beam.sdk.transforms.Gro
浏览 7提问于2019-03-20得票数 0
我从BigQuery中动态地获取了数据列表，并将其转换为所谓的波束列表，使用映射转换计算了它的长度，现在我想使用这个length和list来对更大的PCollection进行partition，然后在各个分区上运行一些转换，从而将操作扇出。我在将length & list传递给分区转换时陷入困境。Beam只允许处理输出PCollections通过ParDo函数侧输入，使用pvalue子模块。我正在寻找一个解决办法来做上述工作。
什么能在这里起作用？
InputsSingletonsCustom PTransformsCustom Side ParDo
with beam.Pipelin
浏览 4提问于2022-10-14得票数 0
我有一个数据流作业，使用DataFlowRunner在本地运行时运行得很好，但当我尝试使用GCP的Composer/AirFlow运行它时，它会给我一个错误：
Exception in thread "main" java.lang.IllegalStateException: Unable to return a default Coder for ConvertToYouTubeMetadata/ParDo(convertToTableRow$1)/ParMultiDo(convertToTableRow$1).output [PCollection]. Correct o
浏览 1提问于2018-07-19得票数 0
回答已采纳
我将向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
回答已采纳
在GCS中，我有一个源CSV文件(没有头)以及头配置CSV文件(只包含列名)。我在Bigquery中还有静态表。我希望通过使用列头映射(config文件)将源文件加载到静态表中。
之前我尝试了不同的方法(我是维护源文件，其中包含头和数据在同一个文件中，然后尝试从源文件中分割头，然后使用头列映射将这些数据插入Bigquery。我注意到这种方法是不可能的，因为数据流将数据洗牌到多个工作节点。所以我放弃了这种方法。
下面的代码我使用了硬编码的列名。我正在寻找从外部配置文件中读取列名的方法(我希望使我的代码成为动态的)。
package com.coe.cog;
import java.io.Buf
浏览 0提问于2018-06-22得票数 1
1回答
Apache模式问题
、、、
我正在尝试执行ETL，其中涉及从HDFS加载文件、应用转换并将它们写入Hive。在使用SqlTransforms通过遵循文档执行转换时，我遇到了以下问题。你能帮忙吗？
java.lang.IllegalStateException: Cannot call getSchema when there is no schema
at org.apache.beam.sdk.values.PCollection.getSchema(PCollection.java:328)
at org.apache.beam.sdk.extensions.sql.impl.schema.BeamP
浏览 0提问于2018-10-25得票数 0
1回答
我正试图把MutationGroups变成SpannerIO的扳手。我们的目标是每10秒编写一次新的MuationGroups，因为我们将使用扳手来查询接近时间的KPI。
当我不使用任何窗口时，我会得到以下错误：
Exception in thread "main" java.lang.IllegalStateException: GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a trigger. Use a Window.into or Window.tr
浏览 1提问于2018-07-23得票数 4
是否有一种方法将侧输入应用于Apache中的BigQueryIO.read()操作。
例如，我在PCollection中有一个值，我想在查询中使用这个值从BigQuery表中获取数据。这有可能使用侧输入吗？还是应该在这种情况下使用其他的东西呢？
我在类似的情况下使用了NestedValueProvider，但我想只有当某个值依赖于我的运行时值时，我们才能使用它。或者我能用同样的东西吗？如果我错了，请纠正我。
我试过的密码：
Bigquery bigQueryClient = start_pipeline.newBigQueryClient(options.as(BigQueryOptions.
浏览 1提问于2017-07-31得票数 4
回答已采纳
这是我编写的Apache：
public class NormalizeTransform
extends PTransform<PCollection<String>, PCollection<SimpleTable>> {
@Override
public PCollection<SimpleTable> expand(PCollection<String> lines) {
ExtractFields extract_i = new ExtractFields();
PCollection<SimpleTab
浏览 2提问于2017-06-05得票数 1
回答已采纳
我正在apache中创建一个管道，在这里我需要用两个键按键群。我有下面的数据集Trasnasctions.txt
101,Credit,100,21-09-2017
101,Dedit,200,22-09-2017
101,Credit,300,23-09-2017
102,Credit,500,26-09-2017
102,Debit,600,27-09-2017
需要使用Account_ID和事务类型对输入数据进行分组。用1键就可以正常工作。
p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
浏览 0提问于2018-02-08得票数 0
我正在构建一个管道，我想看到它的点表示，我用这个
public class PlotDAG extends PTransform<PCollection<KV<Long, ElasticRow>>, PCollection<KV<Long, ElasticRow>>> {
private static final Logger LOG = LogManager.getLogger(PlotDAG.class);
public static PlotDAG of() {
return new Plot
浏览 11提问于2022-11-09得票数 0
2回答
在apache beam中执行Pcollection之后，哪种方法是更好的左连接？
pcoll1 = [('key1', [[('a', 1)],[('b', 2)], [('c', 3)], [('d', 4)],[('e', 5)], [('f', 6)]]), ('key2',[[('a', 12)],[('b', 21)], [('c', 13)]]), ('key3',[[('a'
浏览 0提问于2017-08-05得票数 4
尝试使用BeamSQL取消嵌套PCollection的嵌套类型。让我们假设PCollection包含员工及其详细信息。这里的详细信息在嵌套集合中。因此，如果我们使用像"SELECT PCOLLECTION.details FROM PCOLLECTION"这样的BeamSQL，那么在单独的PCollection中获得嵌套类型的细节作为数组集合。但是，当我想要从嵌套类型集合中获取特定列作为详细信息时，就会出现错误，比如找不到列名。尝试了SQL(类似于BigQuery BeamSQL ) "SELECT X.address FROM PCOLLECTION, Unnest(
浏览 12提问于2019-05-08得票数 0
1回答
当我将消息从pubsubio插入到BigQuery时，我会得到以下错误。
我怎样才能插入从酒吧到烧烤的记录。我们是否可以将pcollection转换为列表，或者是否还有其他替代方案？
AttributeError：'PCollection'对象没有属性'split'
这是我的代码：
def create_record(columns):
#import re
col_value=record_ids.split('|')
col_name=columns.split(",")
for i i
浏览 0提问于2018-06-22得票数 1
我正在尝试在PCollection<Object>上应用SQLTransform。在这里，CustomSource转换在runtime.Hence生成一个Pojo，运行SQLTransform的对象的类型在编译时是未知的。 Pipeline p = Pipeline.create(options);
PCollection<Object> objs = p.apply(new CustomSource());
Schema type = Schema.builder().addInt32Field("c1&#
浏览 46提问于2019-07-04得票数 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
我有一个基本的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
背景
我的束流管道设计用于加工阿夫罗SpecificRecordBase型元件。
为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都有自己的字段：
class Dog extends SpecificRecordBase {
....
}
class Cat extends SpecificRecordBase {
...
}
管道将读取输入Kafka中的元素，处理元素，并将处理后的元素放入输出Kafka中，如下所示：
Pipeline pipeline = Pipeline.create(getOptions());
pipeline.getCoderRegist
浏览 14提问于2020-03-02得票数 0
回答已采纳
我使用beam从包含json对象的文件中准备了一个Pcollection<BeamRecord>对象。
下面的代码解析并将json行映射到ChatHistory对象，然后将映射的对象转换为BeamRecord。最后，我尝试在返回的BeamSql上使用PCollection<BeamRecord>，但是我得到了异常，不能将SerializableCoder转换为BeamRecordCoder。
PCollection<ChatHistory> json_objects = lines.apply(ParDo.of(new ExtractObjectsFn())
浏览 1提问于2018-02-26得票数 1
回答已采纳
我们正在Flink集群上运行Beam应用程序，侧输入为50 of。
基于发送到Kafka中通知主题的通知，侧输入刷新(从外部数据源提取)。
随着应用程序的发展，由于侧输入，全GC频繁发生，每个GC占用30秒左右，暂停任务管理器将心跳发送给主。
在连续心跳丢失后，主假设工人已经死亡，并开始重新分配作业，结果将重新启动应用程序。
我们试着删除侧输入，应用程序可以正常工作。
问题：
Apache束侧输入对侧输入的大小有任何限制吗？
我已经使用asSingleton()创建了侧输入映射，要为每个任务创建单独的副本吗？我给出了15个平行度。它是否要在
浏览 1提问于2020-03-28得票数 1
我的要求是从csv文件中读取数据以及头文件，并使用Python和Dataflow在Google商店中创建相同的结构。我尝试过创建一个示例代码，如下所示。
我的CSV样本在下面，
First Name,Last Name,Date of Birth
Tom,Cruise,"July 3, 1962"
Bruce,Willis,"March 19, 1955"
Morgan,Freeman,"June 1, 1937"
John,Wayne,"May 26, 1907"
我的pyhton2.7代码片段如下所示
import csv
i
浏览 2提问于2020-01-08得票数 0
回答已采纳
我遇到了一个“无法为org.apache.hadoop.hbase.client.Mutation提供编码器”的问题。在FlinkRunner中使用HbaseIO。例外情况如下：
Exception in thread "main" java.lang.IllegalStateException: Unable to return a default Coder for ParDo(HBaseProfile)/ParMultiDo(HBaseProfile).output [PCollection]. Correct one of the following root caus
浏览 0提问于2018-11-08得票数 1
1回答
如何在apache beam中解析apache beam数据并存储在bigquery表中？例如: JSON数据
[{ "name":"stack"},{"id":"100"}].
如何解析JSON数据，并转换为PCollection K，V，将存储在烧烤表？感谢你的帮助！！
浏览 0提问于2018-05-11得票数 0
我使用的子类来实现CombineFn，而不是使用SerializableFunction的实现
package me.examples;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import java.util.HashSet;
import java.util.Set;
public class ConcatWordsCombin
浏览 0提问于2019-06-04得票数 1
回答已采纳
下面的代码主要是从apache文档中复制粘贴的。还是找不出这个错误。同时，我也在努力学习beam。有人能推荐一些好的教程/文章吗？
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.PCollection;
public class reports {
public static v
浏览 7提问于2022-03-05得票数 -1
2回答
我正在尝试使用ZetaSQL解析的SQL转换来运行Beam管道。我从设置选项开始
options.setPlannerName("org.apache.beam.sdk.extensions.sql.zetasql.ZetaSQLQueryPlanner");
当我尝试使用任何给定的查询创建SqlTransform时，我会得到
java.util.ServiceConfigurationError: org.apache.beam.repackaged.sql.com.google.zetasql.ClientChannelProvider: Provider org.apac
浏览 29提问于2019-12-04得票数 0
1回答
我有两个扩展PTransform的类，叫做CompositeCall2和CompositeCall。
我必须首先调用CompositeCall，然后在CompositeCall中完成工作后，我必须在我的数据流程序中调用CompositeCall2。我正在制作我的数据流作业的模板，所以并行处理正在进行，这使得我的工作变得困难。
Code.:Tester.java
package Testing2;
import java.util.List;
import org.apache.beam.runners.dataflow.DataflowRunner;
import org.apache.b
浏览 0提问于2017-10-16得票数 1
1回答
是否可以在pardo(classname(args))中提供参数。因为我试过了，但是它说进程函数需要3个args，而给定的是2个。我也试过init func。没有希望。请帮帮忙。如果可能的话，让我知道如何重写这段代码。注意:分裂和过滤工作良好，因为没有提供are。
import apache_beam as beam
class splitcols(beam.DoFn):
def process(self,elements):
return [elements.split(',')]
class filtercols(beam
浏览 1提问于2019-11-18得票数 0
回答已采纳
如何实现beam sql udf函数将一列拆分为多列？我已经在bigquery udf函数中实现了： CREATE TEMP FUNCTION parseDescription(description STRING)
RETURNS STRUCT<msg STRING, ip STRING, source_region STRING, user_name STRING>
LANGUAGE js AS """
var arr = description.substring(0, description.length - 1).split(","
浏览 27提问于2020-04-16得票数 0
回答已采纳
当我想将我的未绑定输入从Kafka加载到BigQuery时，我遇到了.withMethod()选项。通过使用Method.FILE_LOAD，我还必须指定触发频率以及非零numFileShards。
我的问题是
文件碎片的数量控制多少？它是用来做什么的？在我的观察中，它绝对不是在我的GCS临时位置生成的临时文件，对我来说是可见的。但是我想知道我应该在这里选择几个数字？
根据我引用的源代码，默认值应该是1000，但实际上是0，所以当我没有显式地设置它时，我得到了一个异常，当我将它设置为1时，异常消失了，但是我还是不知道它是什么，我正在设置什么，lol。
/**Control使用
浏览 1提问于2018-09-10得票数 2
回答已采纳
我是Apache的新手，来自PySpark的dataframe。我很难在另一束计算中使用一束计算的输出。基本上，我希望执行一个聚合，该聚合生成一个值(例如平均值)，并将此聚合的结果用作后续聚合中的python原语(例如浮点)。例如：
import apache_beam as beam
DATA = [
beam.Row(val="hello"),
beam.Row(val="stackoverflow,"),
beam.Row(val="plz"),
beam.Row(val="halp"),
]
with
浏览 4提问于2022-04-06得票数 0
回答已采纳
2回答
我希望能够通过自定义键使用分组，但这是我到目前为止的尝试，
我们为KV对象的键使用了自定义类，因为我们希望使用更复杂的条件进行GroupBy，而不是使用字符串等简单的键匹配。
PCollection<KV<Multikey, Iterable<SomeObject>> pc2 =
pc.apply(GroupByKey.<Multikey, SomeObject>create());
使用equals方法表示匹配条件。
class Multikey implements Serializable{
List<String>
浏览 21提问于2019-03-29得票数 3
在Apache中使用MultiOutputReciver到PCollectionTuple和TupleTag有麻烦。
logSchema是我的AvroGenerated类，用于处理传入日志。日期、类型、消息等。我想要做的是将不同类型的日志(错误、警告、通知)存储在不同的PCollections中。
我得到了这个错误java: incompatible types: logSchema cannot be converted to capture#1 of ?
对于processElement内部的每个processElement在branching extends DoFn<logSch
浏览 5提问于2021-11-18得票数 0
我在apache_beam中找到了专门用于tfrecords的io函数，这可以实现：
from apache_beam.io.tfrecordio import ReadFromTFRecord
class VerifyOutput(beam.DoFn):
def process(self, pcollection):
try:
pcollection = pcollection.element
except AttributeError:
pass
logging.info(pcollec
浏览 2提问于2017-09-19得票数 2
回答已采纳
我在试着
1.从Cloud Pub/Sub读取JSON事件
2.每15分钟使用文件加载将事件从Cloud /Sub加载到BigQuery，以节省流插入的成本。
3.目标将根据JSON事件中的"user_id“和"campaign_id”字段而不同，"user_id“将是数据集名称，"campaign_id”将是表名。分区名称来自事件时间戳。
4.所有表的架构保持不变。
我对Java和Beam很陌生。我认为我的代码主要是做我想做的，我只是在这里需要一点帮助。
但是我无法访问JSON消息中的"campaign_id“和"user_id”字段。因此，我
浏览 0提问于2018-04-17得票数 0
回答已采纳
1回答
Apache写入
、、
无法使用Apache Beam JavaSDK写入PubSub。
我正在尝试使用beam从PubSub中读取数据，进行处理，然后将数据写入PubSub主题，但我无法找到如何写入PubSub的工作示例。
是否有人可以帮助适当的转换，以写入PubSub主题。
.apply("Create pubsub messages", ParDo.of(new DoFn<String, PubsubMessage>() {
@DoFn.ProcessElement
public void processElement(ProcessContext c) throws
浏览 0提问于2017-11-14得票数 2
我尝试使用数据流将数据从pubsub流流到Datastore。我搜索了谷歌提供的模板。
注意PubsubToDatastore不起作用。所以，我试着调试这个。
这就是我所做的。
添加errorTag
添加窗口处理(pubsub生成无界数据，而数据存储不能接受无界数据)
添加扁平化(向数据存储写入窗口数据的方法为none。所以，我想开窗了。)
这是我的密码。
package com.google.cloud.teleport.templates;
import com.google.cloud.teleport.templates.common.Datast
浏览 0提问于2018-08-20得票数 0
我用Dataflow运行程序测试了我的代码，但是它返回了一个错误：
> Error message from worker: java.lang.RuntimeException:
> org.apache.beam.sdk.util.UserCodeException:
> com.fasterxml.jackson.core.JsonParseException: Unrecognized token
> 'WindowReiterable[ ]
> org.apache.beam.runners.dataflow.worker.GroupAlsoB
浏览 6提问于2020-03-11得票数 1
我希望有来自Beam (SqlTransform)输出的嵌套行，但是失败了。
问题：
从SqlTransform中用嵌套行输出行的正确方法是什么？(行类型是用描述的，所以我相信它是受支持的)
如果这是一个缺陷/缺失的功能，问题是梁本身？还是依靠跑步？(我目前在DirectRunner上使用，但将来将使用DataflowRunner。)
版本信息：
操作系统: macOS 10.15.7 (Catalina)
Java: 11.0.11 (AdoptOpenJDK)
Beam SDK: 2.32.0
这是我试过的，没有运气。
方解石方言
SELECT ROW(
浏览 5提问于2021-09-18得票数 0
回答已采纳
我的问题非常类似于另一篇文章：。
但是，我在那里尝试了解决方案(将GlobalWindows()应用于侧输入)，但它似乎没有解决我的问题。
我有一个Dataflow管道(但我使用DirectRunner进行调试)和Python，其中的主要输入是来自PubSub的日志，而侧输入是来自基本不变的数据库的关联数据。我想加入这两种方法，使每个日志都与来自相同近似时间的侧输入数据配对。如果没有关联的日志，多余的侧输入可以删除。
我所看到的行为是，管道似乎是作为一个线程运行的。它首先处理所有侧输入元素，然后开始处理主输入元素。如果侧输入有界(非流)，这是很好的，管道可以合并输入并运行到完成。但是，如果侧输
浏览 15提问于2022-05-07得票数 0
回答已采纳
此代码仅以一个文件作为输入，但当我传递:-
D://beam//csv//*.csvor D://beam//csv//20*.csv作为抛出:-的参数
Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.nio.file.InvalidPathException: Illegal char <*> at index 17: D:\\beam\\csv\\20*.csv
at org.apache.beam.runners.direc
浏览 0提问于2020-04-17得票数 0
问题陈述
我们正在将多个csv文件消耗到pcollection中，->应用beam SQL来转换数据，->写入结果的p集合。如果我们在所有源pCollections中都有一些数据，那么这是非常好的，而beam SQL会用一些数据生成新的集合。当Transform pCollection生成空pCollection时，在编写netApp存储网格时，
Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.io.IOException: Failed
浏览 15提问于2022-02-03得票数 0
回答已采纳
我使用Dataflow 1.9 (JAVA )读取Pubsub消息和无缝流到BigQuery中，而不显式地设置TableRow中的每个列。下面是转换的代码片段。
PCollection<TableRow> payloadTableRow = pipeline
.apply("Read",PubsubIO.Read.subscription(***MY_SUBSCRIPTION***)
.withCoder(TableRowJsonCoder.of()));`
上面的代码工作得很好，我可以看到主题中的Pubsub消息被转换为PCollection<
浏览 3提问于2017-08-08得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
