如何将JSON数组反序列化为Apache beam PCollection<javaObject>
Apache Beam是一个用于大规模数据处理的开源框架，它提供了一种统一的编程模型，可以在不同的分布式处理引擎上运行，如Apache Flink、Apache Spark和Google Cloud Dataflow等。在Apache Beam中，PCollection是数据处理的核心概念，它代表了一个不可变的、分布式的数据集合。
要将JSON数组反序列化为Apache Beam的PCollection<javaObject>，可以按照以下步骤进行操作：
- 导入必要的依赖：
- 导入必要的依赖：
- 创建一个Apache Beam的Pipeline对象：
- 创建一个Apache Beam的Pipeline对象：
- 使用TextIO读取包含JSON数组的文件，并将其转换为PCollection<String>：
- 使用TextIO读取包含JSON数组的文件，并将其转换为PCollection<String>：
- 使用JsonToRow将JSON字符串转换为Row对象：
- 使用JsonToRow将JSON字符串转换为Row对象：
- 其中，schema是一个描述JSON结构的Schema对象，可以通过Apache Beam提供的SchemaBuilder来构建。
- 如果需要将Row对象转换为自定义的Java对象，可以使用Apache Beam的ParDo转换：
- 如果需要将Row对象转换为自定义的Java对象，可以使用Apache Beam的ParDo转换：
- 这里的MyJavaObject是自定义的Java对象，根据实际情况进行定义。
- 最后，可以使用TextIO将PCollection<MyJavaObject>写入文件或其他目标：
- 最后，可以使用TextIO将PCollection<MyJavaObject>写入文件或其他目标：
这样，就完成了将JSON数组反序列化为Apache Beam的PCollection<javaObject>的过程。在实际应用中，可以根据具体需求进行进一步的数据处理和分析。
腾讯云相关产品和产品介绍链接地址：
相关·内容
这是我编写的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
回答已采纳
我使用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
回答已采纳
我正在自学Apache Beam，专门用于解析JSON。我能够创建一个简单的示例，将JSON解析为POJO，将POJO解析为CSV。它要求我对我的简单POJO类使用.setCoder()。 pipeline
.apply("Read source JSON file.", TextIO.read().from(options.getInput()))
.apply("Parse to POJO matching schema", ParseJsons.of(Person.class))
浏览 48提问于2021-10-31得票数 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
回答已采纳
尝试使用BeamSQL取消嵌套PCollection的嵌套类型。让我们假设PCollection包含员工及其详细信息。这里的详细信息在嵌套集合中。因此，如果我们使用像"SELECT PCOLLECTION.details FROM PCOLLECTION"这样的BeamSQL，那么在单独的PCollection中获得嵌套类型的细节作为数组集合。但是，当我想要从嵌套类型集合中获取特定列作为详细信息时，就会出现错误，比如找不到列名。尝试了SQL(类似于BigQuery BeamSQL ) "SELECT X.address FROM PCOLLECTION, Unnest(
浏览 12提问于2019-05-08得票数 0
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
我使用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
1回答
如何在apache beam中解析apache beam数据并存储在bigquery表中？例如: JSON数据
[{ "name":"stack"},{"id":"100"}].
如何解析JSON数据，并转换为PCollection K，V，将存储在烧烤表？感谢你的帮助！！
浏览 0提问于2018-05-11得票数 0
我有一个在json文件中描述的表，在此基础上，我希望稍后将集合创建为sideInput。
PCollection<KV<Boolean, Map<String, Object>>> pC = p_jsonstring
.apply("create ...", MapElements.via( (String input) -> {
try {
O
浏览 1提问于2017-06-19得票数 2
我有一个基本的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
使用下面的窗口函数，
Window.<KV<String, Long>>into(FixedWindows.of(Duration.standardDays(1)))
.triggering(
AfterWatermark.pastEndOfWindow()
.withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.s
浏览 0提问于2020-05-07得票数 1
背景
我的束流管道设计用于加工阿夫罗SpecificRecordBase型元件。
为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都有自己的字段：
class Dog extends SpecificRecordBase {
....
}
class Cat extends SpecificRecordBase {
...
}
管道将读取输入Kafka中的元素，处理元素，并将处理后的元素放入输出Kafka中，如下所示：
Pipeline pipeline = Pipeline.create(getOptions());
pipeline.getCoderRegist
浏览 14提问于2020-03-02得票数 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
回答已采纳
使用“数据流作业”从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
我尝试使用数据流将数据从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
我们正在创建一个数据流管道，我们将从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
回答已采纳
如何实现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
回答已采纳
我遇到了一个“无法为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
我使用的子类来实现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
回答已采纳
2回答
我想读取一个csv文件，并将其写入到BigQuery使用阿帕奇光束数据流。为此，我需要将数据以字典的形式呈现给BigQuery。如何使用apache beam转换数据以实现此目的？
我的输入csv文件有两列，我想在BigQuery中创建一个后续的两列的表。我知道如何在BigQuery中创建数据，这很简单，我不知道的是如何将csv转换成字典。下面的代码是不正确的，但应该给出了我想要做什么的想法。
# Standard imports
import apache_beam as beam
# Create a pipeline executing on a direct runner (local
浏览 1提问于2016-12-16得票数 9
回答已采纳
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
使用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
回答已采纳
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
下面的代码主要是从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
我试着熟悉Apache beam Kafka IO，并得到以下错误 Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: org.apache.kafka.common.errors.SerializationException: Size of data received by LongDeserializer is not 8
at org.apache.beam.runners.direct.DirectRunner$DirectPipelineRe
浏览 18提问于2020-07-03得票数 0
回答已采纳
我正在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
我将向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
回答已采纳
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
我正在尝试在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
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
1回答
是否可以使用Beam SQL执行基于列的查找？我遇到了一个类BeamJoinTransforms.JoinAsLookup，但找不到任何可用的代码片段。目前，为了在Apache梁代码中执行查找，我遵循了一种做法，即使用CoGroupByKey/SideInput执行“左连接”，并通过在代码中维护列映射来生成过滤后的TableRows。我相信这也可以通过使用Beam来实现，但要遵循一种更有效的方式来处理查找。有没有人有这方面的工作片段？寻找示例，其中我可以转换为Pcollection<Row>的PCollection<TableRow>和执行字段查找使用Beam SQL库
浏览 2提问于2018-08-10得票数 0
1回答
尝试使用中的BigQuery表获得唯一值。使用Group子句实现了BeamSQL中的条件(下面的示例查询)。其中一列具有浮动数据类型。执行作业时出现以下异常，
由: org.apache.beam.sdk.coders.Coder$NonDeterministicException: org.apache.beam.sdk.coders.RowCoder@81d6d10引起的不是确定性的，因为:所有字段都必须具有确定性编码。由: org.apache.beam.sdk.coders.Coder$NonDeterministicException: FloatCoder引起的不是确定性的，因为
浏览 0提问于2019-05-29得票数 1
回答已采纳
我有以下代码：
PCollection<KV<String, Cell>> first = ...;
PCollection<String> lines = first
.apply("Build lines", Combine.<String, Cell, String>perKey(new MergeCellsFn()))
;
Cell看起来像这样：
public class Cell {
public final int index;
public final String value;
publ
浏览 0提问于2020-08-22得票数 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
回答已采纳
问题陈述
我们正在将多个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
回答已采纳
1回答
我很难理解Apache Beam中的连接(例如)是否可以连接整个行。
例如：
我有两个CSV格式的数据集，其中第一行是列标题。
第一个：
a,b,c,d
1,2,3,4
5,6,7,8
1,2,5,4
第二个：
c,d,e,f
3,4,9,10
我想在c列和d列上进行左连接，这样我就可以得到：
a,b,c,d,e,f
1,2,3,4,9,10
5,6,7,8,,
1,2,5,4,,
然而，所有关于Apache Beam的文档似乎都说，在连接时，PCollection对象需要是KV<K, V>类型，所以我将PCollection对象分解为KV<String, String>
浏览 31提问于2018-06-06得票数 1
回答已采纳
我正在使用apache beam向gcs编写一个json。但是遇到了以下错误 NotImplementedError: offset: 0, whence: 0, position: 50547, last: 50547 [while running 'Writing new data to gcs/write data gcs/Write/WriteImpl/WriteBundles/WriteBundles'] 不知道为什么会发生此错误。相同的代码如下： class WriteDataGCS(beam.PTransform):
"""
浏览 20提问于2019-09-11得票数 0
回答已采纳
1回答
我在apache beam中构建了一个管道，我对此很好奇，将ptransform应用到一个列表和一个pcollection之间有什么区别，性能是受此影响的还是仅仅是pcollection是不可变的，这是用apache beam来获取管道的一个坏方法吗？
浏览 4提问于2022-04-02得票数 0
回答已采纳
2回答
我有一些地图格式的数据，我想使用波束管道将它们转换为tfrecords。下面是我编写代码的尝试。我已经在python中尝试了这一点，但是我需要在java中实现它，因为有一些业务逻辑是我无法移植到python的。在这个中可以找到相应的工作python实现。
import com.google.protobuf.ByteString;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
import org.apache.beam.sdk.io.TFRecor
浏览 8提问于2020-04-17得票数 0
回答已采纳
由于某种原因，每次我尝试在管道中创建模式件PCollection (Apache )时，使用beam.Select()或beam.Row()并尝试在Dataflow上运行它时，我都会得到一个致命的错误，它会导致工作人员在VM上死亡。奇怪的是，当我使用在本地运行时，不会得到这个错误。
我想确保这不是我在我的自定义代码中犯的愚蠢错误。我试过使用Google的示例，例如下面简化的代码片段：
import json
import logging
import apache_beam as beam
from apache_beam.options.pipeline_options impo
浏览 2提问于2021-07-26得票数 0
我在Apache Beam中构建了一个小函数来执行查找/联接:给定一个集合映射A到B，另一个集合映射B到C，返回一个集合映射A到C。
class Main {
private static <A,B,C> PCollection<KV<A,C>> lookup(
PCollection<KV<A,B>> collection,
PCollection<KV<B,C>> lookup
){
var leftTag = new Tupl
浏览 11提问于2018-08-02得票数 0
回答已采纳
public static void main(String[] args) {
//Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setRunner(DataflowRunn
浏览 4提问于2017-12-08得票数 0
回答已采纳
在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的新手，我正在尝试开发一个管道。我有2个相同格式的pCollections，我还有另一个其他格式的pCollection。我试着做一个ParDo函数，对于pCollection 3中的每个项目，根据这个项目的值，搜索项目是否存在于pCollection 1或2中，以完成pCollection 1或2的信息输出。但我不知道这个ParDo函数是如何做到的。
这是我的代码：
output = (
pCollection1, pCollection2, pCollection3
| 'ParDo function' >> b
浏览 0提问于2020-09-21得票数 3
我正在构建一个管道，我想看到它的点表示，我用这个
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
我将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
回答已采纳
我用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
我是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
回答已采纳
在FlatMapElements和MapElements中使用"into“方法时，我得到了错误。请帮助我解决这个问题，因为我得到了“TypeDescriptor(FlatMapElements)类型未定义的方法”错误
对于上面的TypeDescriptor，它没有定义，但是不能替代任何东西，而且我是Apache的新手。
救命啊！
下面是源代码代码：
import org.apache.beam.examples.common.ExampleUtils;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.
浏览 0提问于2020-02-18得票数 0
1回答
使用Apache，我正在进行计算--如果它们成功，我希望将输出写入一个接收器，如果出现故障，我希望将其写入另一个接收器。
有任何方法来处理Apache中的元数据或基于内容的路由吗？
我广泛地使用了Apache，因此在我的脑海中，基于先前转换的结果，我应该使用 (可能由我在消息头上设置的元数据标志确定)将消息路由到另一个接收器。Apache是否具有类似的功能，或者我是否只需要一个顺序的转换来检查PCollection并在转换中处理写入接收器？
理想情况下，我希望这个逻辑(为试图清晰而详细地写成)
result = my_pcollections | 'compute_stuff'
浏览 0提问于2018-11-08得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
