Python中的Apache Beam -我们可以写一个文件脚注吗
Apache Beam是一个用于大规模数据处理的开源框架，它提供了一种统一的编程模型，可以在不同的分布式处理引擎上运行，包括Apache Flink、Apache Spark和Google Cloud Dataflow等。在Python中使用Apache Beam可以实现高效的数据处理和分析。
Apache Beam的主要特点包括：
- 统一的编程模型：Apache Beam提供了一种统一的编程模型，可以在不同的分布式处理引擎上运行，无需修改代码。这使得开发人员可以专注于业务逻辑而不用关心底层的分布式处理细节。
- 扩展性和灵活性：Apache Beam支持水平扩展，可以处理大规模的数据集。同时，它还提供了丰富的转换操作和窗口操作，可以灵活地处理不同类型的数据。
- 容错性和可靠性：Apache Beam具有容错性和可靠性，可以处理在分布式环境中出现的故障和错误。它提供了数据重试、故障恢复和数据一致性等机制，确保数据处理的可靠性。
- 支持多种数据源和格式：Apache Beam支持多种数据源和格式，包括文件、数据库、消息队列和流式数据等。它提供了丰富的输入和输出连接器，可以方便地与各种数据源进行交互。
Apache Beam在以下场景中具有广泛的应用：
- 批量数据处理：Apache Beam可以高效地处理大规模的批量数据，例如数据清洗、数据转换和数据分析等任务。
- 流式数据处理：Apache Beam支持流式数据处理，可以实时地处理数据流，例如实时监控、实时分析和实时推荐等场景。
- 批流一体：Apache Beam提供了批流一体的编程模型，可以同时处理批量数据和流式数据，实现更灵活的数据处理。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云数据计算服务（Tencent Cloud Data Compute，TDC）：腾讯云提供的大数据计算服务，支持Apache Beam等开源框架，帮助用户快速构建和运行大规模数据处理任务。详细信息请参考：腾讯云数据计算服务
- 腾讯云流计算Oceanus（Tencent Cloud Stream Compute，Oceanus）：腾讯云提供的流式数据处理服务，支持Apache Beam等开源框架，可以实时处理和分析数据流。详细信息请参考：腾讯云流计算Oceanus
总结：Apache Beam是一个用于大规模数据处理的开源框架，它提供了统一的编程模型，支持批量数据处理和流式数据处理。在Python中使用Apache Beam可以实现高效的数据处理和分析。腾讯云提供了相应的数据计算服务和流计算服务，可以帮助用户快速构建和运行数据处理任务。
相关·内容
在运行wordcount示例管道时，我一直在跟踪并得到一个错误：
..。文件"apache_beam/io/fileio.py"，第281行，在glob返回gcsio.GcsIO().glob(路径，限制) AttributeError：'NoneType‘对象没有属性'GcsIO’
我试过用我自己的管道，结果也是一样。我不知道这里的问题是什么，因为我认为我完全遵循了本教程，并且这个错误似乎与读/写转换有关。
回溯(最近一次调用)：文件"/System/Library/Frameworks/Python.framework/Versions/2
浏览 6提问于2017-02-26得票数 2
回答已采纳
我正在使用virtualenvwrapper来管理一个项目，在这个项目中，我只是暂时运行它
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
with beam.Pipeline(options=PipelineOptions()) as p:
pass
这将返回以下错误
Traceback (most recent call last):
File "path/to/pipeline.py", line 1, in <mod
浏览 10提问于2020-08-20得票数 1
1回答
我正在尝试使用Google Cloud Dataflow (Python SDK)将Google PubSub消息写入Google Cloud Storage。消息以json格式进入PubSub，我必须定义一个模式，以便将它们写入Google Cloud Storage中的parquet格式。
根据其他用户的建议，我从特别研究和源代码开始着手这项任务。
第一个不是我想要做的，因为它将更改应用到json文件(它通过一个窗口合并它们，将原始的json放入一个字段"message“中，并添加一个表示发布时间的时间戳)。
第二个源代码(源代码)更适合这个用例。具体地说，模式是从BigQuery
浏览 3提问于2020-07-23得票数 0
2回答
我使用来自的代码
并将其改为：
def getFullTableName(pn,tn):
return "{0}:{1}".format(pn,tn)
...
(
pipeline | "Read Data From Input Topic" >> beam.io.ReadFromPubSub(topic=data_topic)
| "Get Table data from input row" >> beam.Map(lambda r : data_ingestion.getData(r
浏览 12提问于2022-01-05得票数 0
我正在尝试使用下面的代码读取Bigquery中的一些数据和文件系统中的一些数据。
( beam.io.Read(beam.io.BigQuerySource(query=apn_query，()( beam.combiners.ToList() beam.combiners.ToList() preprocess_rows =p= beam.io.ReadFromText(file_path，coder=UnicodeCoder())
但是，当我运行这个管道时，我会出现以下错误
文件"/etl/dataflow/etlTXLPreprocessor.py"，第125行，在ru
浏览 4提问于2020-03-01得票数 1
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
回答已采纳
我们有一个从Google Cloud Datastore到Bigquery的数据传输问题。我们需要在python中为这项工作创建数据流脚本。此作业应使用python中的管道将数据从数据存储传输到bigquery。对于python中的这项工作，它需要"Apache Beam“library.But Apache Beam库不工作。有人能帮我们吗？
浏览 0提问于2016-10-07得票数 0
无法为流运行数据流和获取未读消息的计数--我需要一个流数据流来监视未读消息--我的代码中有任何问题。
import logging
import datetime
import argparse
import yaml
import apache_beam as beam
from apache_beam import Pipeline, ParDo, DoFn
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import monitoring_v3
from goog
浏览 4提问于2021-06-10得票数 0
如果Google NLP API在文本标识的语言不受支持时返回错误，我在管道上的一次迭代中遇到错误。我在管道内处理此异常时遇到了问题，以利用apache beam的高效bach请求。 features = nlp.types.AnnotateTextRequest.Features(
extract_entities=True,
extract_document_sentiment=True,
extract_syntax=False
)
p = beam.Pipeline()
(p
| beam.Create(message_log_df['messa
浏览 76提问于2021-10-17得票数 0
在我新的虚拟环境里。
我跑
pip search apache-beam
我得到了
apache-beam (2.9.0)
然后我跑了
pip install apache-beam
pip list
但是我安装了apache-beam2.2，而不是2.9
apache-beam 2.2.0
然后我跑
python -m apache_beam.examples.wordcount --output cout
我得到了错误
The Apache Beam SDK for Python is supported only on Python 2.7.
从这份文件
beam 2.9将
浏览 0提问于2019-01-27得票数 0
回答已采纳
我希望这里有人能帮忙。我一直在谷歌这个错误像疯了，但没有发现任何东西。
我有一个管道在本地执行时工作得很好，但是在GCP上执行时它会失败。下面是我得到的错误消息。
工作流失败。原因:S03:写入转换fn/WriteMetadata/ResolveBeamFutures/CreateSingleton/Read+Write转换fn/WriteMetadata/ResolveBeamFutures/ResolveFutures/Do+Write转换fn/WriteMetadata/WriteMetadata。一个工作项尝试了4次，但没有成功。每次工人最终失去与服务的联系。该工作项目是在以下方面尝
浏览 0提问于2018-03-29得票数 0
回答已采纳
2回答
目前是否可以使用Apache从python中的gzip文件中读取？我的管道正在用以下代码从gcs中提取gzip文件：
beam.io.Read(beam.io.TextFileSource('gs://bucket/file.gz', compression_type='GZIP'))
但是我发现了一个错误：
UnicodeDecodeError: 'utf8' codec can't decode byte 0x8b in position 1: invalid start byte
我们注意到，在python源代码中，压缩文件似乎是
浏览 9提问于2016-08-22得票数 6
回答已采纳
2回答
我用python (在GCP Dataflow上)做了一个简单的管道，用来读PubSub和写大查询，但不能处理管道上的异常来创建替代流。 在一个简单的WriteToBigQuery示例中： output = json_output | 'Write to BigQuery' >> beam.io.WriteToBigQuery('some-project:dataset.table_name') 我试图将其放入try/except代码中，但它不起作用，因为当它失败时，异常似乎被抛出到我的python执行之外的Java层： INFO:root:201
浏览 14提问于2019-01-30得票数 5
回答已采纳
2回答
我是Apache的新手，刚刚开始使用Python。关于Apache，我知道管道、收集、转换、ParDo和DoFn的高水平。
在我当前的项目中，管道已经实现，使用熊猫使用下面提到的语法读取、转换和写入文件。
我想了解如果这是正确的实现Apache，因为我们是直接读写文件只使用熊猫，而不是处理每个元素的文件元素。
步骤：
创造管道
创建输入文件路径的pcollection
调用DoFn并传递文件路径
在DoFn (读、转换和写)中使用熊猫做所有的事情。
高级代码示例：
import **required libraries
class ActionClass(beam.D
浏览 0提问于2019-09-16得票数 4
我们正在尝试使用Dataflow将数据从BigQuery上传到Microsoft服务器。我们使用我安装了pyodbc和msodbcsql18的Dataflow flex环境。
当在交互式会话中使用DirectRunner在对接器容器中运行我们的数据流脚本时，我们没有问题:读取BigQuery、转换数据并最终上传到server。
但是，当在交互式会话中使用DataflowRunner运行相同的脚本时，我们会得到以下错误：
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/ap
浏览 8提问于2022-09-27得票数 0
回答已采纳
2回答
我是Apache的新手，我尝试完成三项任务。
从表中阅读前30项
从桌子上阅读前30位商店
从bigquery中选择必需列，并对列Item、和Stores应用筛选器。
我有下面的代码，以执行管道
with beam.Pipeline(options=pipeline_args) as p:
#read the dataset from bigquery
query_top_30_items = (
p
| 'GetTopItemNumbers' >> beam.io
浏览 0提问于2021-03-18得票数 0
回答已采纳
当我尝试通过将文档写入mongodb时 | "Write User Doc to Mongo" >> beam.io.WriteToMongoDB(uri=MONGO_URI,
db="dbname",
coll="col_name"
浏览 30提问于2019-09-04得票数 0
回答已采纳
1回答
我遵循了以下关于在本地设置google-cloud-ml的说明: MAC/LINUX
但是，在使用以下命令验证安装时，我会得到以下错误
curl https://raw.githubusercontent.com/GoogleCloudPlatform/cloudml-samples/master/tools/check_environment.py | python
追溯(最近一次调用)：文件""，第70行，文件"/Users/pratyusha/miniconda2/envs/cloudml/lib/python2.7/site-packages/google/
浏览 4提问于2017-02-28得票数 0
2回答
当我从DirectRunner切换到Dataflowrunner：Pardo显然不起作用时，我遇到了问题。当我将runner设置为Dataflowrunner时，def process(self，query)从未运行。我可以看到这个工作在GCP上工作，但是，我的方法InsertPostgresql不使用Dataflowrunner。
根据日志错误，很明显，ParDo没有识别出“psycopg2 2”。
process NameError:在处理上述异常时未定义名称“psycopg2”
我想知道为什么？
def run_pipeline():
os.enviro
浏览 4提问于2020-11-11得票数 1
回答已采纳
我正在运行Apache管道，从读取文本文件，对这些文件执行一些解析，并将解析后的数据写入Bigquery。
为了保持解析的简短性，我的代码忽略了解析和google_cloud_options，我的代码如下：(Apache-Bam2.5.0带有GCP加载项和数据流作为运行程序)
p = Pipeline(options=options)
lines = p | 'read from file' >>
beam.io.ReadFromText('some_gcs_bucket_path*') | \
'parse xml to d
浏览 0提问于2018-07-30得票数 4
class Mp3_to_npyFn(beam.DoFn):
def process(self, element):
filename, e = element
# get mp3 from the storage
bucket = storage.Client().get_bucket('BUCKET_NAME')
blob = bucket.get_blob(filename)
tmp_mp3 = TemporaryFile()
blob.download_to_fil
浏览 0提问于2019-05-08得票数 0
回答已采纳
我跟踪，希望在读取文本文件时访问文件名和内容。我有GCP存储中的文本文件，(以压缩格式- gzip)，当我试图读取这些文件时，它给我的错误如下：
Error message from worker: Traceback (most recent call last):
File "apache_beam/runners/common.py", line 1417, in apache_beam.runners.common.DoFnRunner.process
File "apache_beam/runners/common.py", line 624,
浏览 14提问于2022-11-03得票数 0
用例
我希望解析来自云存储的多个文件，并将结果插入到BigQuery表中。
选择一个要读取的特定文件可以正常工作。但是，当我使用* glob模式将一个文件切换到包含所有文件时，我感到很困难。
我执行的任务如下：
python batch.py --project foobar --job_name foobar-metrics --runner DataflowRunner --staging_location gs://foobar-staging/dataflow --temp_location gs://foobar-staging/dataflow_temp --output foob
浏览 2提问于2017-03-29得票数 3
回答已采纳
我在Python2.7上创建了一个管道，它运行在上。当我从我的笔记本电脑在本地部署它时，这个管道工作得很好。我现在希望通过CloudBuild来部署它。这是我的cloudbuild.yaml文件：
steps:
- name: "docker.io/library/python:2.7"
args: ["pip", "install", "-t", "/workspace/lib", "-r", "requirements.txt"]
- name: "doc
浏览 1提问于2020-01-27得票数 0
回答已采纳
我正在尝试使用PortableRunner通过Docker在Apache上运行计数示例。
使用./gradlew -p sdks/python/container docker构建SDK工具容器。但是，当对创建的映像执行对接时，会出现以下错误：
使用默认标记:来自守护进程的最新错误响应: Get apache.bintray.io/v2/: x509:证书对*.bintray.io、bintray.io有效，而不是$userId-docker- apache.bintray.io
使用./gradlew beam-runners-flink_2.11-job-server:r
浏览 0提问于2019-05-09得票数 0
我正在尝试在Apache光束中运行一个非常简单的程序来测试它是如何工作的。 import apache_beam as beam
class Split(beam.DoFn):
def process(self, element):
return element
with beam.Pipeline() as p:
rows = (p | beam.io.ReadAllFromText(
"input.csv") | beam.ParDo(Split())) 在运行这段代码时，我得到以下错误 .... some more
浏览 23提问于2019-03-05得票数 2
回答已采纳
我正在尝试为Google Cloud Dataflow创建自己的模板，以便可以从GUI执行作业，从而使其他人更容易执行。我遵循了教程，创建了我自己的PipelineOptions类，并用parser.add_value_provider_argument()方法填充它。然后，当我尝试使用my_options.argname.get()将这些参数传递到管道中时，我得到一个错误，告诉我该项不是从运行时上下文中调用的。我不明白这个。args不是定义流水线图形本身的一部分，它们只是输入文件名、输出表名称等参数。
下面是代码。如果我硬编码输入文件名、输出表名称、写部署和分隔符，它就能工作。如果我用它们的
浏览 3提问于2018-07-11得票数 2
在本地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
我只想知道，在python的新版本(3.X)中，我们是否有更多的管道I/O和运行时参数。如果我是正确的，那么当前apache beam只提供基于文件的IOs: textio，avroio，tfrecordio当使用python时。但是对于Java，我们有更多的选择，比如基于文件的IOs、BigQueryIO、BigtableIO、PubSubIO和SpannerIO。
在我的需求中，我想使用Python3.x在GCP数据流管道中使用BigQueryIO，但目前还不能使用。有没有人有一些关于ETA的更新，它什么时候才能由apache beam提供。
浏览 0提问于2020-02-10得票数 1
我试图按照下面的链接中解释的设计实现一个用例，但是遇到了错误。任何指点都会有很大帮助。
用例说明步骤：
从PubSub获取流原始事件。
验证接收到的原始事件。
过滤特定类型的事件。
创建过滤事件的字典。
同时，通过窗口操作传递过滤后的事件并对其进行聚合。
2种类型的输出-原始事件字典，聚合事件字典。
按照上述链接中解释的设计，原始事件字典属于LowUrgency类别，聚合事件属于HighUrgency类别。
对原始事件尝试“FILE_LOADS”方法，以避免成本部分。
尝试聚合事件的“STREAMING_INSERT”方法，因为它需要实时可用。
下
浏览 0提问于2020-08-07得票数 2
我正在处理多个文件输入，并希望将它们输入到Dataflow管道中。但是，我要保留产出，使之与投入的数量相匹配。让我们假设我们有三个不同的文件：
gcs_files = ['gs://bucket/<file1_dir>', 'gs://bucket/<file2_dir>', 'gs://bucket/<file3_dir>']
我们希望将这些文件输入到一个简单的读->写管道中，如下所示：
import apache_beam as beam
from apache_beam.io import R
浏览 32提问于2022-07-13得票数 0
回答已采纳
在beam 2.14.0下使用以下代码 | "FixedWindow" >> beam.WindowInto(beam.window.FixedWindows(4 * 60),
trigger=beam.trigger.Repeatedly(
beam.trigger.AfterProcessingTime(delay=1 * 60)
),
accumulation_
浏览 30提问于2019-08-27得票数 3
我可以在Java SDK文档中看到，我们可以在FileIO.ReadableFile实用程序类- https://beam.apache.org/releases/javadoc/2.2.0/org/apache/beam/sdk/io/FileIO.ReadableFile.html#open--中指定压缩 但是，我使用的是Python，它可以作为参数(apache_beam.io.fileio.ReadMatches(compression=None, skip_directories=True))使用，但是浏览一下源代码，我认为它没有任何作用-- https://beam.apache
浏览 20提问于2019-12-16得票数 0
回答已采纳
我需要处理存储在中的定制二进制文件。
为此，我编写了一个自定义。如文档所述，它由定义为Java glob、单个文件或单个文件的偏移范围的文件模式支持。
在我的例子中，我需要使用带有几个特定文件名的Java，比如这个/path/{file1,file1,file3}。当我在本地文件系统上测试它时，它可以正常工作，但是如果我将它与Google (gs://bucket/{file1,file2,file3})一起使用，它将无法找到任何文件，并得到以下堆栈跟踪：
java.io.IOException: Error executing batch GCS request
at org
浏览 5提问于2017-10-27得票数 2
2回答
旁瓣静态数据
、、
在ParDo中处理数据时，我需要使用存储在上的JSON模式。我觉得这可能是副业？我阅读了他们称为文档()的页面，其中包含了一些关于apache_beam.pvalue.AsSingleton和apache_beam.pvalue.AsSideInput的内容，但是如果我在Google上搜索这些文档的使用情况，结果将为零，而且我也找不到任何关于Python的例子。
如何从ParDo中的存储中读取文件？或者我是在ParDo之前旁听我的管道，但是如何利用ParDo的第二个源呢？
编辑
我的主要数据来自BQ：beam.io.Read(beam.io.BigQuerySource(...
侧输入也来自B
浏览 7提问于2019-12-23得票数 2
回答已采纳
1回答
摘要:一些本地包可以工作，而有些则不能
我的波束应用程序的结构：
-setup.py
-app/__init__.py
-app/main.py
-package1/__init__.py
-package1/one.py
-package2/__init__.py
-package2/two.py
-package3/__init__.py
-package3/three.py
在main.py中：
from package1 import one
from package2 import two
from package3 import three
在setup.py中
impor
浏览 0提问于2018-09-27得票数 2
3回答
我编写代码将CSV文件中的数据注入到谷歌的BigQuery中。我用阿帕奇光束做管道。 这是管道代码： list_of_data = open_file()
DistrictAllocationAndListStore_data = (p
| 'CreateDictData from DistrictAllocationAndListStore File' >> beam.Create(list_of_data)
| 'Renam
浏览 19提问于2021-08-03得票数 1
回答已采纳
我开始学习apache beam，在尝试构建我的第一个管道(用python)时，我遇到了一个奇怪的行为。 以下是我的代码中的元素： 一个文件：count_words.txt，其中包含： ?Strawberry
?Carrot
?Eggplant
?Tomato
?Potato 代码的一个有效版本：wordcount_exercise.py import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
input_file = "data/input/count_
浏览 22提问于2021-08-09得票数 1
回答已采纳
我的问题很简单:我想使用Python将s3中的一个拼图文件读入Apache中的PCollection中。
我知道apache_beam.io.parquetio模块，但是这个模块似乎不能直接从s3读取(或者它是？)。
我知道apache_beam.io.aws.s3io模块，但这个模块似乎返回了一个s3文件对象，或者某种不是PCollection的东西(或者是吗？)。
那么最好的方法是什么呢？
浏览 2提问于2022-08-08得票数 0
1回答
我正在创建一个数据管道，使用Apache beam从google存储桶(GCS)中获取XML文件作为输入，并将其转换为JSON文件。我尝试使用python的'xmltodict‘库首先将XML转换为python dict，然后使用python json.dumps()函数将其转换为json格式。我已经为光束管道的每个步骤创建了单独的beam.DoFn类。
我在一个小文件(小于1MB )上测试了流水线，它工作正常。代码在directrunner (少于1分钟)和dataflow runner (5-6分钟，包括启动和停止数据流作业)上运行。但是，当我使用dataflow runner处理
浏览 8提问于2019-10-25得票数 0
最近，由于超出请求大小，我的数据流作业从BigQuery应用程序接口抛出HttpBadRequestError。
Error message from worker: generic::unknown: Traceback (most recent call last):
File "apache_beam/runners/common.py", line 1246, in apache_beam.runners.common.DoFnRunner._invoke_bundle_method
File "apache_beam/runners/commo
浏览 20提问于2021-02-23得票数 0
回答已采纳
我有一个.txt文件，它有一些数据碎片。我想要将文件读取、操作和重组为json格式，我如何使用Apache在Python中做到这一点？
txt文件是这样的。
IDX- |00000016325.00|000000000500|000000000500|D|000000006385|00000014400.00|000000004600|00000014425.00|000000000600|000000000000|\'c7\uc0\u8800
IDX- |00000007100.00|000000000800|000000000800|D|000000006386|000000066
浏览 3提问于2022-03-26得票数 1
我的Apache beam管道实现了自定义转换和ParDo的python模块，这些模块进一步导入了我编写的其他模块。在Local runner上，这工作得很好，因为所有可用的文件都在相同的路径中。在使用Dataflow runner的情况下，管道失败，并出现模块导入错误。
如何使所有数据流工作者都可以使用自定义模块？请给我建议。
下面是一个示例：
ImportError: No module named DataAggregation
at find_class (/usr/lib/python2.7/pickle.py:1130)
at find_class (/usr/l
浏览 0提问于2018-07-10得票数 7
我正在尝试遵循Apache贡献指南
使用Python 2.7.15步骤：
git克隆git@github.com:apache/beam.git
创建一个新的虚拟env (--无侧包)
pip安装cython
在这个sdk/python目录上运行pip install '.[gcp,test]'
产出：
pip install '.[gcp,test]'
Processing /Users/jcc/beam/sdks/python
Complete output from command python setup.py
浏览 0提问于2018-08-07得票数 0
回答已采纳
1回答
我对apache_beam.transforms.deduplicate.Deduplicate转换有一个问题。请看下面的代码示例：
with beam.Pipeline() as pipeline:
(
pipeline
# | 'Load' >> beam.Create(['a', 'b', 'b']) ## <- works fine
| 'Load' >> beam.io.ReadF
浏览 22提问于2022-10-27得票数 0
回答已采纳
这是我的进口代码
from __future__ import absolute_import
import datetime
import json
import logging
import re
import apache_beam as beam
from apache_beam import combiners
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
from apache_beam.io.gcp.datastore.v1.datastoreio import ReadFromDat
浏览 5提问于2017-10-20得票数 2
回答已采纳
1回答
触发云存储-数据流
、、、、
我现在开始工作，我需要一些帮助，我有一个自定义模型，我使用apache beam创建了一个管道，该管道从存储桶内的文件夹中获取来自csv文件的数据，然后将数据抛到bigquery表中，这个表已经按照我的要求工作，但是由于它是一个批处理管道，它只在我运行数据流时运行，所以我想自动化这个函数，条件是当加载一个新文件时，这个作业自行运行，我怎么做呢？
模板数据流
import apache_beam as beam
import os
from apache_beam.options.pipeline_options import PipelineOptions
pipeline_options
浏览 7提问于2022-04-28得票数 1
我正在尝试在云数据流上运行apache-beam管道。原来的函数已经被部署为云函数，它应该创建一个读取文本文件并插入到大型查询中的数据流作业。但它无法在数据流上运行。函数和误差如下所示。
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options import StandardOptions
import a
浏览 20提问于2020-02-16得票数 2
我对传输和读取流数据非常陌生，所以我希望我的问题不是太琐碎。
在将数据写入其他文件之前，我将使用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
我有一个简单的数据流管道，并试图从云shell执行，
代码：
from __future__ import print_function
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
with beam.Pipeline(options=PipelineOptions()) as p:
lines = p | 'Read' >> beam.io.ReadFromText('test.csv')
li
浏览 0提问于2019-03-30得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
