我试着从GCS批量处理.avro文件，并将结果作为拼图文件写回GCS，数据是时间序列，元素是时间戳。如何从拼图输出中的timestamp列创建列索引？在Pandas/Dask中，它是一个简单的.set_index('timestamp')语句。class AddTimestampDoFn(beam.DoFn):
def process(self, element
浏览 13提问于2021-08-10得票数 1
回答已采纳
我正在使用Apache Beam Python SDK，并且我正在尝试使用apache_beam.io.parquetio从拼图文件中读取数据，但我还想将文件名(或路径)添加到数据中，因为它也包含数据。我查看了建议的模式，并阅读到Parquetio类似于fileio，但它似乎没有实现允许遍历文件并将其添加到派对中的功能。
有没有人想出一个好办法来
浏览 18提问于2021-02-28得票数 0
回答已采纳
我想读取一个GZIP压缩的帕奎特文件从GCS到BigQuery使用Python SDK for Apache光束。但是，apache_beam.io.parquetio.ReadFromParquet方法似乎不支持从压缩文件中读取。根据源代码，压缩类型被硬编码为UNCOMPRESSED。有没有一个技巧来读取压缩的拼图文件，而不需要在GCS中预先解压缩文件？如果这是唯一的方法，有没有办法在GCS中直接
浏览 2提问于2019-11-25得票数 0
1回答
在我的数据流(beam)工作流中，我使用来自Python的datetime包(在gcp上使用jupyter notebook )。当我用pip安装缺少的包时，一切都很好。现在我想在gcp上运行我的转换为数据流作业。我总是得到错误的"NAME ERROR“，因为datetime是未知的(当我使用Apache beam SDK for Python时得到这个
浏览 1提问于2020-11-16得票数 0
1回答
消息以json格式进入PubSub，我必须定义一个模式，以便将它们写入Google Cloud Storage中的parquet格式。
根据其他用户的建议，我从特别研究和源代码开始着手这项任务。第一个不是我想要做的，因为它将更改应用到json文件(它通过一个窗口合并它们，将原始的json放入一个字段"message“中，并添加一个表示发布时间的时间戳)。具体地说，模式是从BigQuery中的表中提取的数据自
浏览 3提问于2020-07-23得票数 0
我正试图在Flink Runner上运行Python光束作业。我已经在我的本地机器上启动了一个minikube集群，并转发了8081端口来查看localhost:8081上的Flink Dashboard。部署的flink集群上的所有东西似乎都在运行，但当我尝试执行基本管道(在DirectRunner上运行良好)时，我在Flink completed作业仪表板上得到了这个堆栈跟踪错误： Job failedduring initialization of JobManager
org.apache
浏览 530提问于2021-02-04得票数 0
假设我们想要转换shakespeare文件夹中的每个文件，并且处理比计算字数要密集得多。 在不手动指定不同分支的情况下，这样的事情在同一管道中是可能的吗？不是这个 # This processes all the files in the same "branch" in Dataflow.
p | beam.Create([/shakespeare/*
浏览 13提问于2021-07-06得票数 0
我有一个简单的管道，3周前它可以正常工作，但是我已经返回代码来增强它，当我试图运行代码时，它返回了以下错误：我正在Google平台上通过Cloud通过简单地执行Python3 <dataflow.py>
浏览 6提问于2020-10-23得票数 4
如何实现beam sql udf函数将一列拆分为多列？我试图在beam udf函数中返回一个对象，但是beam sql似乎不支持object.*语法。我还试图返回一个映射或数组，但仍然出错。有没有办法在beam中实现同样的udf？我尝试使用MapElement方法，但得到了错误，似乎输出行需要与输入行相同的模式，例如： import org.apache.beam
浏览 27提问于2020-04-16得票数 0
回答已采纳
我开始学习apache beam，在尝试构建我的第一个管道(用python)时，我遇到了一个奇怪的行为。 以下是我的代码中的元素： 一个文件：count_words.txt，其中包含： ?Potato 代码的一个有效版本：wordcount_exercise.py import apache_beam as beam
from apache_beam</e
浏览 22提问于2021-08-09得票数 1
回答已采纳
1回答
我们有一个apache数据流作业，它读取来自Big的数据，在使用Datastax驱动程序将数据写入Cassandra之前将它们转换为POJO。最近，我向表中添加了一个新的blob列，并将一个ByteBuffer字段添加到POJO中。at org.apache.beam.sdk.Pipeline.run(Pipeline.java:311)
at org.apache.beam</em
浏览 0提问于2019-01-31得票数 0
回答已采纳
然而，这一次，我更加不清楚应该如何读取/调试输出日志：ERROR: Could not find a version that satisfies the requirement在GKE上，这是我的输出：[server] File "/app/shared/to_db.py", line/python</
浏览 9提问于2021-11-24得票数 0
我的最终目标是创建一个自定义的管道，所以“发布/订阅到云存储”模板对我来说是不够的，而且我对Java一无所知，这让我开始在Python中进行调整。这是我目前所得到的(Apache Beam Python SDK 2.10.0)： import apache_beam as beam
TOPIC_PATH="projects/<my-projec
浏览 6提问于2019-02-18得票数 7
回答已采纳
我正在尝试运行由本地Flink集群支持的Apache，以便使用卡夫卡主题，如中所述。终端输出
2.23.0: Pulling from apache/beam_java_sdk/beam_java_sdk:2.23.0 docker.io/
浏览 1提问于2020-08-30得票数 0
回答已采纳
我看到BigQuery有一个内置的I/O连接器，但我们的很多数据都存储在Snowflake中。有没有连接到Snowflake的解决方法？我能想到的唯一办法就是使用sqlalchemy运行查询，然后将输出转储到云存储存储桶中，然后Apache-Beam就可以从存储在存储桶中的文件中获取输入数据。
浏览 22提问于2020-01-29得票数 3
回答已采纳
我只想知道，在python的新版本(3.X)中，我们是否有更多的管道I/O和运行时参数。如果我是正确的，那么当前apache beam只提供基于文件的IOs: textio，avroio，tfrecordio当使用python时。但是对于Java，我们有更多的选择，比如基于文件的IOs、BigQueryIO、BigtableIO、PubSubIO和SpannerIO。在我的</em
浏览 0提问于2020-02-10得票数 1
假设我有emp文本文件(enum，ename，edept，data.the )包含在文件的中间，index=2键下面。d10,IT根据梁的行为，我认为关键应该引入到第一列，以比较其他文件。我可以把它带到第一栏，但我不能合并。这是我的密码。() )
if __name__ == "__main__":当我运行代码时，我得到的错误如下
浏览 4提问于2021-12-25得票数 1
回答已采纳
"test1": "test1", } }
} 我的输出是这样的PubSub的JSON中有多少test_units来扁平化数据，并在单个dict中返回那么多行。return rows 下面是Stackdriver日志中的错误： Error processing instruc
浏览 18提问于2019-04-18得票数 1
回答已采纳
1回答
LE: TL;DR；如何在Python中创建无界数据源？有可能吗？我正在构建一个流数据流，它将继续处理来自具有时间戳、id和读取值的传感器的浮点值，将值放在FixedWindows中2秒，然后输出一个聚合。" >> beam.ParDo(DummyWindowPrint())我遇到的问题是，GroupByKey之后的操作只有在输入完成后才会启动，但是我希望在WindowInto上使用一个触发器>
浏览 1提问于2020-03-12得票数 2
在python数据流作业中，我每天都会收到这个错误。
10毫秒，而没有输出或完成在sun.misc.Unsafe.park(本地方法)的完成。CompletableFuture.java:1729) at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895) at org.apache</em
浏览 6提问于2020-02-04得票数 0
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM
腾讯云开发者
