1回答
当前的波束管道正在使用FileIO.matchAll().continuously()以流的形式读取文件。这将返回PCollection。我想将这些具有相同名称的文件写回另一个gcs bucket，也就是说，每个PCollection都是一个文件metadata/readableFile，它应该在经过一些处理后写回另一个bucket。有没有什么接收器可以用来将每个PCollection项目写回GCS，或者有什么方法可以做到？是否有可能为每个元素创建一个窗口，然后使用一些GCS接
浏览 6提问于2018-01-13得票数 1
我见过很多Apache Beam的例子，你从PubSub读取数据并将其写入GCS bucket，但是有没有使用KafkaIO并将其写入GCS bucket的例子呢？例如我的用例是将数据从Kafka流式传输到GCS存储桶，所以如果有人在GCP中提出了一些更好的方法，也很受欢迎。
谢谢。问候你，阿南特。
浏览 12提问于2020-03-17得票数 1
2回答
如何标记数据流中的已读文件？
、、、、
我正在使用数据流从GCS存储桶中读取文件，并对其进行一些转换。为此，我使用了beam.io.ReadFromText()方法。
浏览 5提问于2020-07-02得票数 1
4回答
我的GCS 上有~300 GCS文件，最终我试图将它导入BigQuery，但是它有一些错误的数据结构(我是mongoexport从MongoDB获得的)
字段名"$date“无效。因此，现在我的方法是从GCS逐行读取源文件，并使用python将每一行处理后的代码上传到BigQuery。
浏览 5提问于2018-10-09得票数 3
回答已采纳
1回答
Apache写入
、、
无法使用Apache Beam JavaSDK写入PubSub。我正在尝试使用beam从PubSub中读取数据，进行处理，然后将数据写入PubSub主题，但我无法找到如何写入PubSub的工作示例。
浏览 0提问于2017-11-14得票数 2
1回答
我正在使用以下片段的稍微调整的版本将PubSub消息写入GCS： def __init__(self, output_path, prefix订阅每秒获得500个元素，因此每秒有3-10个元素无法跟上。(window.FixedWindows(self.window_size, 0))
| "Decode windowed elements" >> beam<
浏览 5提问于2022-06-23得票数 1
回答已采纳
我在Apache上工作非常新，在这里我尝试编写一个管道，从Google BigQuery中提取数据，并使用Python格式将数据写入GCS。使用beam.io.read(beam.io.BigQuerySource())，我可以从BigQuery读取数据，但不确定如何以CSV格式将其写入GCS。import logging
f
浏览 1提问于2018-10-22得票数 6
回答已采纳
1回答
用apache_beam.io.filesystems.FileSystems如何用ParDo和DoFn写GCS ?？我已经从pardo获得了csv格式的输出，我是否需要编写另一个pardo来将其写入gcs，或者我是否可以直接导入模块以将其直接写入gcs？请帮帮忙
浏览 24提问于2019-03-16得票数 0
回答已采纳
我正在尝试从pandas数据帧获取输入到apache beam管道，并将其写入GCS。在不使用dataflow/apache beam的情况下，我能够在GCS中写入数据帧数据。def database_to_gcs(self, type='full'): with open(self.tablemetada
浏览 1提问于2019-04-03得票数 1
我已经在一个码头上安装了apache_beam Python和airflow。Python版本: 3.5我正在尝试使用**DataflowPythonOperator**执行apache波束管道。当我从气流用户界面运行DAG时，我得到
导入错误:导入apache_beam作为光束。当我尝试使用Python2和airflow 1.10.3进行相同的对接设置时，两个月前，操作符没有返回任何错误，而是按预期工作。在SSH进入码
浏览 1提问于2019-09-06得票数 1
我希望通过Python在GCS中摄取和编写Avro文件。通过Avro利用Python，这是目前可能的吗？如果是这样的话，我怎么做呢？我在消息来源中看到了TODO对此的评论，所以我不太乐观。
浏览 4提问于2016-06-13得票数 1
回答已采纳
我在google数据流上运行apache beam管道。它从GCS桶中读取数据，并在处理后写入GCS桶。这条管道处理日本数据。在堆栈驱动程序日志中，日语字符正在正确显示。因此，我主要是在想，或者是在向GCS写入数据时，编码器没有设置，或者我们必须对GCS文件格式进行更改。解决这个问题需要帮助。
我尝试过在束流管道中设置编码。此外，在运行pipleine时，我尝试使用parma编码运行jar。import org.apache<
浏览 5提问于2019-10-22得票数 0
1回答
我有一个从BigQuery获取数据并将其写入GCS的管道，但是，如果我发现任何拒绝，我想将它们正确地放到Bigquery表中。我正在将拒绝收集到全局列表变量中，然后将列表加载到BigQuery表中。当我使用dataflowrunner运行它时，它并不保证顺序(我希望pipeline1在pipeline2之前运行。在Dataflow中有使用python的依赖管道吗？| 'get data' >> beam.io.Read(beam.io.BigQuerySo
浏览 5提问于2020-09-19得票数 1
回答已采纳
我有一个简单的管道，3周前它可以正常工作，但是我已经返回代码来增强它，当我试图运行代码时，它返回了以下错误：import apache_beam as beamfrom apach
浏览 6提问于2020-10-23得票数 4
at org.apache.beam.runners.dataflow.util.DataflowTemplateJob.getJobId(DataflowTemplateJob.java:41)at org.apache.beam.runners.dataflow.DataflowPipelineJob.getJobWithRetries(DataflowPipelineJob.java:559)
at org.apache.beam.runners.dataflow.Data
浏览 10提问于2022-09-19得票数 1
我正在尝试编写一个管道来读取来自pubsub的流，并使用google cloud dataflow和apache stream将其写入bigquery。我有这样的代码：from apache_beam.transforms.window import FixedWindows
topic = 'gcs_path = "gs://???&q
浏览 1提问于2017-11-29得票数 0
2回答
我正在尝试使用Apache从Google获取文件。我试过了with beam.Pipeline() as pipeline:lines = (pipeline | beam.Create(filenames))这将返回一个类似于PCollection[[19]: Create/Map(de
浏览 9提问于2022-11-12得票数 0
我试图从GCS桶中读取XML文件的集合，并处理它们，其中集合中的每个元素都是一个字符串，表示整个文件，但我找不到一个关于如何实现这一点的好例子，也无法从主要关于Java版本的Apache文档中理解它。_executor.await_completion()但是现在我看到了这种方法给了我从每个文件中的一行作为管道元素
浏览 6提问于2018-08-22得票数 3
回答已采纳
1回答
我有一个GCS，在这里我每分钟都能得到文件，我使用ApacheBeampythonsdk.i创建了一个用于输入gcs桶和输出gcs bucket.my数据流的发布/子主题创建了一个流数据流，但是我的输出没有存储在输出import enums from datetime import datetime
浏览 2提问于2019-03-07得票数 3
回答已采纳
我在Google Dataflow中运行了一个Apache beam流水线。它从Kafka中读取数据并将其流式插入到Bigquery。(BatchedStreamingWrite.java:374)at org.apache.beam.ru
浏览 7提问于2021-06-01得票数 5
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
