在Beam中读写序列化的协议
是指在Apache Beam框架中用于读取和写入数据的序列化协议。序列化是将数据转换为字节流的过程，以便在网络上传输或存储在磁盘上。Beam框架支持多种序列化协议，包括Avro、Parquet、JSON、CSV等。
- Avro：Avro是一种数据序列化系统，具有动态模式定义和强大的数据结构演化能力。它支持多种编程语言，并且具有高效的压缩和快速的序列化/反序列化速度。在Beam中使用Avro可以实现高效的数据读写操作。腾讯云相关产品：腾讯云数据万象（https://cloud.tencent.com/product/ci）。
- Parquet：Parquet是一种列式存储格式，适用于大规模数据处理。它具有高效的压缩率和查询性能，并且支持谓词下推和列剪裁等优化技术。在Beam中使用Parquet可以实现高效的数据读写和查询操作。腾讯云相关产品：腾讯云数据仓库（https://cloud.tencent.com/product/dw）。
- JSON：JSON是一种轻量级的数据交换格式，易于阅读和编写。在Beam中使用JSON可以方便地处理结构化数据，并且与多种编程语言兼容。腾讯云相关产品：腾讯云云数据库Redis版（https://cloud.tencent.com/product/redis）。
- CSV：CSV是一种常用的文本格式，用于存储表格数据。在Beam中使用CSV可以方便地处理扁平化的数据，并且与多种工具和系统兼容。腾讯云相关产品：腾讯云云数据库CDB版（https://cloud.tencent.com/product/cdb）。
这些序列化协议在Beam中具有不同的特点和适用场景，开发人员可以根据实际需求选择合适的协议来读写序列化数据。
相关·内容
我正试图用我的自定义对象创建一个虚拟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回答
我们有一个apache数据流作业，它读取来自Big的数据，在使用Datastax驱动程序将数据写入Cassandra之前将它们转换为POJO。最近，我向表中添加了一个新的blob列，并将一个ByteBuffer字段添加到POJO中。
我是如何创建ByteBuffer的
String str = objectMapper.writeValueAsString(installSkuAttributes);
byte[] bytes = str.getBytes( StandardCharsets.UTF_8 );
pojo.setInstallAttributes(ByteBuffer.wrap(
浏览 0提问于2019-01-31得票数 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
回答已采纳
1回答
我编写了一个代码来使用Apache来执行一个LeftOuterJoin -- apache提供的类可以轻松地工作--提供一个连接类LeftOuterJoin，而当我使用POJO类或字符串、Integer，以KV格式长，但是当我在KV中使用TableRow并抛出一个异常时，整个代码都能正常工作。我也分享了下面的代码，以供参考.
Apr 12, 2018 6:26:03 PM org.apache.beam.sdk.Pipeline validate
WARNING: The following transforms do not have stable unique names: ParDo(A
浏览 0提问于2018-04-12得票数 2
下载eclipse中的maven依赖项
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>0.2.0-incubating</version></dependency>
在将gs路径更改为C:// //misc.txt.Get后下载并运行示例时，我没有传递任何runner.How来传递运行器选项和输出参数，同时运行eclipse？
浏览 0提问于2016-10-27得票数 0
回答已采纳
1回答
我们有一个步骤，在Dataflow的newline上拆分Pubsub消息。我们有一个通过代码的测试，但它似乎在生产中失败了。看起来我们一次在多个地方得到相同的Pubsub消息(至少据我所知)。
我们应该用另一种方式写第一次考试吗？或者，这仅仅是一个关于在Apache中不做什么的艰难教训吗？
import apache_beam as beam
from apache_beam.io import PubsubMessage
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.ut
浏览 2提问于2021-09-01得票数 1
回答已采纳
1回答
我已经创建了一个用于在数据流上运行的beam管道p，并且希望在运行管道之前将一些东西写入文件中。我的代码是：
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
import time
pipeline_options = PipelineOptions(runner='DirectRunner')
pipeline_opt
浏览 0提问于2019-03-01得票数 0
回答已采纳
1回答
使用Apache，注册的编码器不能工作。
我想在BigQuery的SimpleFunction中使用TableSchema，但是它需要序列化。我将TableSchemaCoder添加到CodeRegistry中，但它似乎没有被使用。
我该怎么解决呢？
// Coder
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import org.apache.beam.sdk.coders.AtomicC
浏览 8提问于2017-10-25得票数 1
回答已采纳
1回答
我有一个Apache管道，它从DynamoDB读取数据。要读取数据，我使用Apache SDK。我需要在用例中读取特定/筛选数据，这意味着我必须在filterExpression中使用DynamoDBIO。我现在的代码如下，
Map<String, AttributeValue> expressionAttributeValues = new HashMap<>();
expressionAttributeValues.put(":message", AttributeValue.builder().s("Ping").build());
浏览 4提问于2022-11-01得票数 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
回答已采纳
我正在用Python语言在Apache Beam上构建一个管道，并使用Dataflow上的笔记本进行原型设计。在尝试加载JSON时，我意识到我在beam.io.ReadFromText(file_pattern, coder=JsonCoder())中使用的JSON编码器(基本上是JSON.loads())不能规范化JSON。这意味着集合中的一些列只是以字符串的形式嵌套的JSON。
因此，为了规范化JSON，我决定使用JsonPickle。但是，即使在安装并重新启动内核之后，模块导入也不起作用。你能帮帮忙吗？此外，还有一种更好的方法在Apache Beam世界中规范化JSON，请分享。
浏览 30提问于2020-05-14得票数 1
使用“数据流作业”从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
我有一个现有的带有Java 8、ApacheBam2.27.0、Maven和Dagger 2的Apache项目。
我在Kotlin : Kotlin 8中迁移了这个项目，版本为1.5.0。
我使用了1.5.0版本的Kotlin，因为1.4.3与Beam和Maven插件有问题(无法阅读类: VirtualFile：)
除了使用带有Typedescriptor和lambda表达式的本机MapElement或FlatMapElement之外，一切似乎都很好。
我的pom.xml文件的一部分
<properties>
<beam.version>2.27.0&l
浏览 4提问于2021-04-30得票数 0
回答已采纳
我正在使用Apache光束，并将数据写入BigQuery。我的流水线在intellij中使用Direct runner在本地工作得很好，并且我能够在intellij中本地写入BigQuery表。然而，一旦我在Spark Cluster上部署了代码，我就得到了异常"java.lang.IllegalArgumentException: Invalid lambda反序列化“。 User class threw exception: java.lang.IllegalArgumentException: unable to deserialize Custom DoFn With Exe
浏览 144提问于2021-10-20得票数 0
回答已采纳
我有一个这样简单的课程：
class Beam
{
public string Name { get; set; }
public double Width { get; set; }
public double Height { get; set; }
}
我用它作为Dictionary中的一个值
var addresses = new Dictionary<string, Beam>
{
{"Beam1", new Beam{Name = "B1", Width = 10, Height = 10}},
{
浏览 4提问于2014-03-29得票数 2
回答已采纳
我正在尝试使用apache的WriteToKafka类(python )编写一个流到Kafka主题。但是，它无休止地运行脚本(没有错误)，并且不向主题写入流。我必须取消运行，它不会停止，不会产生错误。任何帮助都是非常感谢的。下面可以找到一个很小的例子来重现问题
from typing import Tuple
import os
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.kafka import WriteToK
浏览 7提问于2022-07-06得票数 0
我试图在DataBricks笔记本上使用Apache运行一个简单的管道，但是我无法创建任何自定义函数。下面是一个简单的例子：
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
def my_func(s):
print(s)
pipeline_options = PipelineOptions([
"--runner=DirectRunner",
])
with beam.Pipeline(options=pipeline_
浏览 7提问于2022-08-16得票数 0
我试图使用DynamicDestinations写入BigQuery中的分区表，其中分区名为mytable$yyyyMMdd。如果我绕过了动态目的地，并在.to()中提供了一个硬编码的表名，那么它就能工作；但是，对于动态目的地，我得到了以下例外：
java.lang.IllegalArgumentException: unable to serialize org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1@6fff253c
at org.apache.beam.sdk.util.SerializableUtils.serializeToByt
浏览 8提问于2017-09-11得票数 1
我正在尝试编写一个光束流管道，它简单地从PubSub队列中读取数据，解析数据并写入两个BigQuery表中的任何一个。因此，代码利用副输出从DoFn中写入两个表中的一个。我遇到以下错误消息: java.lang.IllegalArgumentException: unable to serialize DoFnAndMainOutput{doFn=com.pipeline.PubSubToBigQuery$ParsePubSubMessage@50eca7c6，mainOutputTag=Tag}。我将完整的错误消息、DoFn和测试类附加到下面： DoFn： public static cl
浏览 10提问于2019-01-14得票数 1
回答已采纳
我正在尝试设置一个数据流作业，用于将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
回答已采纳
给定已知的TCP端口和远程beam.smp服务的名称，以及已知的cookie，是否有可能短路握手阶段，并直接与目标beam.smp服务建立Erlang？
该协议记录在这里：
在这里：
但是，我不清楚是否通过或绑定到特定端口的beam.smp服务来进行recv_ clear /send_clear身份验证。
谢谢您抽时间见我。
浏览 6提问于2013-10-18得票数 5
回答已采纳
我们有一个Android应用程序，它使用Android将数据写入NFC标记。在sdk 29中拒绝Beam之后，是否有其他方法可以使用NFC标记进行数据传输？
是否有可能在NFC标签上使用Google附近的共享？
我没有找到任何文件。
浏览 3提问于2021-04-20得票数 0
回答已采纳
我对传输和读取流数据非常陌生，所以我希望我的问题不是太琐碎。
在将数据写入其他文件之前，我将使用beam Python SDK从PubSub读取数据。因为我收到的数据总是相同的格式，所以我尝试使用特性来解析我从PubSub收到的数据。
接收到的数据总是一个字典name: "my_name", value: 42，所以我的管道看起来像这样：
import typing
import apache_beam as beam
from apache_beam.io import ReadFromPubSub
class MySchema(typing.NamedTuple):
浏览 0提问于2021-07-22得票数 0
我一直在使用Go Beam (v2.13.0)，但无法让在GCP数据流上工作。它在尝试启动org.apache.beam.runners.dataflow.worker.DataflowRunnerHarness时进入崩溃循环。当使用Direct runner在本地运行时，该示例可以正确执行。
该示例完全没有从上面给出的原始示例进行修改。
堆栈跟踪为：
org.apache.beam.vendor.grpc.v1p13p1.com.google.protobuf.InvalidProtocolBufferException: Protocol message had invalid UTF-8
浏览 167提问于2019-07-13得票数 1
在本地Apache实例上使用SqlTransform时，我会收到一个错误。下面是一个简单的测试：
with beam.Pipeline() as p:
pc = (p | beam.Create([
FruitRecipe("pie", "strawberry", 3, 1.5),
FruitRecipe("muffin", "blueberry", 2, 2.),
])
| beam.Map(lambda x: beam.Row(recipe = x[0],
浏览 9提问于2021-01-26得票数 0
当我试图发送或接收一个文件，我一直得到这个错误写在下面。我试图找出调试的原因，每次都停在这一行：
ObjectInputStream Inputs = new ObjectInputStream (socket.getInputStream ());
以下是在输入键盘文件路径后出现的运行错误：
java.io.EOFException
at java.io.ObjectInputStream$PeekInputStream.readFully (ObjectInputStream.java:2353)
at java.io.ObjectInputStream$BlockDataInpu
浏览 4提问于2017-02-14得票数 1
我遇到了一个“无法为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
通过前台NDEF推送API级别10到13，我可以在两个非Android设备之间模拟吗？
我有一个Android和一个黑莓NFC功能的设备。如何在API级别10中在这些设备之间进行仿真？
package com.app.app.nfctag;
import java.nio.charset.Charset;
import android.app.Activity;
import android.content.Intent;
import android.nfc.NdefMessage;
import android.nfc.NdefRecord;
import android.nfc.Nf
浏览 0提问于2012-07-24得票数 1
3回答
在运行数据流作业时，我得到了以下错误。我正在尝试将我现有的beam版本更新为2.11.0，但是在运行时，我会出现以下错误。
org.apache.beam.model.pipeline.v1.RunnerApi$StandardPTransforms$Primitives类没有实现请求的接口com.google.protobuf.ProtocolMessageEnum在org.apache.beam.runners.core.construction.BeamUrns.getUrn(BeamUrns.java:27) at org.apache.beam.runners.core.const
浏览 0提问于2019-03-13得票数 4
1回答
如我们所知，在java中使用Serialization，引用博客的话。
将对象的状态转换为字节流，可以持久化到磁盘/文件中，也可以通过网络发送到运行Java虚拟机的任何其他机器。
REST案例：
现在考虑第二种情况，将一个网络发送到另一个正在运行的jvm，如果我们考虑一个Rest 的例子，即“：port/path/”。
通常，我使用Spring的@RequestMapping将资源模型pojo类作为ResponseEntity返回。我Serializable 没有在模型类中实现接口，而且一切都很好，我在json中得到了API的响应。
ModelX.java
public c
浏览 0提问于2019-07-19得票数 5
目前，我在Python中有一个apache-beam管道，在该管道中，我正在读取拼花，将其转换为dataframe来进行一些熊猫的清理，然后将其转换回我想要编写文件的parquet。看起来是这样的：
with beam.Pipeline(options=pipeline_options) as p:
dataframes = p \
| 'Read' >> beam.io.ReadFromParquetBatched(known_args.input) \
| 'Convert to pandas' >&g
浏览 8提问于2022-10-24得票数 2
回答已采纳
我有一个PCollection，其中每个元素都是一个键，值如下所示：(key, (value1,..,value_n) )
我需要将这个PCollection分成两个处理分支。
像往常一样，我需要整个管道尽可能快，并使用尽可能小的冲压。
我想到了两个想法：
选项1:将PColl拆分为具有多个输出的DoFn
class SplitInTwo(beam.DoFn):
def process(self, kvpair):
key, values = kvpair
yield beam.TaggedOutput('left', (ke
浏览 5提问于2020-12-04得票数 0
回答已采纳
我很难将apache管道编译成数据流模板。我使用自定义aws凭据提供程序类从GCP秘密管理器加载基本aws凭据，并在构建模板时看到以下异常：
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Unexpected IOException (of type java.io.IOException): Failed to serialize and deserialize property 'awsCredentialsProvider' with value 'org.example.iot.b
浏览 15提问于2022-02-25得票数 2
当运行我的代码时，在python或colaborator中出现以下错误，我输入了apache中的所有库，有人在一瞬间给出了这个错误，或者知道它。
用途: aaa_users_py.py -h --流式-beam_ BEAM_SERVICES -类型_check_ TYPE_CHECK_ADDITIONAL
-no_管线_type_check-性能_运行时_type_check-允许不安全_触发器-直接_runner_bundle_重复DIRECT_RUNNER_BUNDLE_REPEAT
-标签标签-转换_名称_映射TRANSFORM_NAME_MAPPING -数据流_kms_key DA
浏览 9提问于2022-06-02得票数 0
回答已采纳
这是我编写的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
回答已采纳
prototxt文件可编译吗？这些文件真的与.proto文件相同吗？
我读过和这个，也读过，但是它没有描述如何处理prototxt文件。
使用：protoc --proto_path=. --java_out=. test.proto我得到一个test.java文件。
但是，protoc --proto_path=. --java_out=. test.prototxt (有效的 test.prototxt文件，而不仅仅是重命名的.proto文件)不会编译。我错过了什么，为什么这里没有提到这件事？
浏览 1提问于2018-04-18得票数 1
1回答
我目前正在尝试在Nexus7上实现一个基本的NFC应用。
我已经能够使用Beam SDK传输NDEF记录(Web Url)，但我被“点击到beam”UI卡住了，就像这里的许多其他人一样。
我知道没有办法禁用这个屏幕或自动按下它，但是，有没有其他可能的解决办法？
例如，一位同事刚刚提出了一些建议，比如将Nexus7安装在tag writer mode中，然后将其他应用程序安装到tag read mode中。
或者我说Android中的NFC只能用于读/写NFC标签，也可以使用beam共享项目，这是正确的吗？
浏览 0提问于2012-10-30得票数 0
回答已采纳
1回答
我试图在中运行一个Apache应用程序，它使用Apache作为运行时。管道使用连接器。我正在尝试，因为Kinesis不允许导出环境变量，导出环境变量似乎不是一种选择。
我试图使用下面的代码进行身份验证。
GoogleCredentials credential = GoogleCredentials
.fromStream(credentialJsonInputStream)
.createScoped("https://www.googleapis.com/auth/cloud-platform", "https
浏览 4提问于2021-08-13得票数 1
2回答
我们正在使用Apache和DirectRunner作为通道来构建一个管道。我们目前正在尝试一条简单的管道，以便：
从Google /Sub中提取数据(目前使用模拟器在本地运行)
反序列化为Java对象
使用固定窗口1分钟的窗口事件
使用自定义CombineFn将这些窗口组合起来，将它们从事件转换为事件列表。
管道代码：
pipeline
.apply(PubsubIO.<String>read().topic(options.getTopic()).withCoder(StringUtf8Coder.of()))
.apply("ParseEve
浏览 2提问于2017-05-16得票数 9
回答已采纳
1回答
UI在GCP数据流上有一个python流管道，它从PubSub中读取数千条消息，如下所示：
with beam.Pipeline(options=pipeline_options) as p:
lines = p | "read" >> ReadFromPubSub(topic=str(job_options.inputTopic))
lines = lines | "decode" >> beam.Map(decode_message)
lines = lines | "Parse&#
浏览 3提问于2020-06-17得票数 2
回答已采纳
我正在构建一个读取Avro通用记录的管道。要在阶段之间传递GenericRecord，我需要注册AvroCoder。文档指出，如果我使用泛型记录，模式参数可以是任意的：
但是，当我将一个空模式传递给AvroCoder.of(Class, Schema)方法时，它会在运行时抛出异常。有没有一种方法可以为GenericRecord创建不需要模式的AvroCoder？在我的例子中，每个GenericRecord都有一个嵌入的模式。
异常和堆栈跟踪：
Exception in thread "main" java.lang.NullPointerException
at org.apa
浏览 2提问于2018-12-13得票数 7
背景
我的束流管道设计用于加工阿夫罗SpecificRecordBase型元件。
为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都有自己的字段：
class Dog extends SpecificRecordBase {
....
}
class Cat extends SpecificRecordBase {
...
}
管道将读取输入Kafka中的元素，处理元素，并将处理后的元素放入输出Kafka中，如下所示：
Pipeline pipeline = Pipeline.create(getOptions());
pipeline.getCoderRegist
浏览 14提问于2020-03-02得票数 0
回答已采纳
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
我在我的苹果电脑上使用DirectRunner测试我的数据流流水线，得到了很多这样的“警告”消息，我可能知道如何摆脱它们，因为它太多了，以至于我甚至看不到我的调试消息。
谢谢
Apr 05, 2018 2:14:48 PM org.apache.beam.sdk.util.MutationDetectors$CodedValueMutationDetector verifyUnmodifiedThrowingCheckedExceptions
WARNING: Coder of type class org.apache.beam.sdk.coders.SerializableCoder ha
浏览 3提问于2018-04-06得票数 6
我正在尝试理解Thrift为javascript自动生成的.read(输入)和.write(输出)函数。
我找不到任何关于如何与这些作为Javascript对象一部分的函数进行交互的文档。
链接还是描述？
浏览 1提问于2012-08-25得票数 1
1回答
用于模式的更好的Jason编码器
、、、、
我收到警告
警告: Jason.Encoder协议已经合并，任何实现都没有效果。如果您想在编译后或测试期间实现协议，请查看Kernel.defprotocol/2 lib/encoder.ex:1文档中的“整合”部分
/Users/dev/projects/haitracker.com/haitracker-be/_build/dev/lib/jason/ebin/Elixir.Jason.Encoder.Any.beam)警告:重新定义模块Jason.Encoder.Any (当前版本从 lib/encoder.ex:1加载)
使用以下编码器
defimpl Jason.Encoder, f
浏览 1提问于2018-09-02得票数 0
回答已采纳
我正在尝试从一桶谷歌云平台编写一个csv到数据存储，其中包含法语字符/口音，但我有一条关于解码的错误消息。
在尝试了从“拉丁语-1”到"utf-8“的编码和解码，但没有成功(使用unicode、unicodedata和编解码器)之后，我尝试手动更改.
我使用的Os默认使用"ascii“编码，我在”Anaconda3 3/envs/py27/lib/site.py“中手动将其更改为utf-8。
def setencoding():
"""Set the string encoding used by the Unicode implementati
浏览 1提问于2018-10-17得票数 3
我正在尝试使用数据流将一些数据从pub sub移动到云存储。我需要向scio / beam提供一个时间戳，这样它就可以将数据分组到窗口中。
我有一个简单的case类来模拟我的事件，它看起来像这样(删除了一些字段)
case class DataEvent(source: String,
record: AnyRef,
timestampUtc: DateTime,
publishedUtc: DateTime)
我的管道是这样
浏览 0提问于2017-09-18得票数 2
我是Apache beam的新手，我使用Apache beam，在GCP.I中作为运行程序使用Dataflow，在执行管道时会出现以下错误。
coder of type class org.apache.beam.sdk.coders.ListCoder has a #structuralValue method which does not return true when the encoding of the elements is equal. Element [Person [businessDay=01042020, departmentId=101, endTime=2020-0
浏览 1提问于2020-06-12得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
