使用python在apache beam中根据条件调用特定的pubsub主题
在Apache Beam中使用Python根据条件调用特定的PubSub主题，您可以按照以下步骤进行操作：
- 首先，确保您已经安装了Apache Beam和相关的Python库。您可以通过以下命令安装Apache Beam和PubSub库：
- 首先，确保您已经安装了Apache Beam和相关的Python库。您可以通过以下命令安装Apache Beam和PubSub库：
- 导入必要的库和模块：
- 导入必要的库和模块：
- 创建一个PubSub客户端实例：
- 创建一个PubSub客户端实例：
- 定义一个条件函数，用于判断消息应该发布到哪个主题：
- 定义一个条件函数，用于判断消息应该发布到哪个主题：
- 创建一个PubSub转换器函数，用于将输入的消息转换为PubSub消息对象：
- 创建一个PubSub转换器函数，用于将输入的消息转换为PubSub消息对象：
- 创建一个Apache Beam管道，读取输入消息并根据条件调用特定的PubSub主题：
- 创建一个Apache Beam管道，读取输入消息并根据条件调用特定的PubSub主题：
- 在上述代码中，您需要将
<project-id>替换为您的项目ID，input-subscription替换为您的输入消息订阅名称。 - 运行Apache Beam管道：
- 运行Apache Beam管道：
通过以上步骤，您可以使用Python在Apache Beam中根据条件调用特定的PubSub主题。根据具体的业务需求和条件，您可以自定义条件函数和主题，并将代码集成到您的云计算解决方案中。
对于腾讯云相关产品和产品介绍链接地址，由于要求不提及特定品牌商，这里无法提供具体链接。但您可以在腾讯云官方网站或文档中查询相关产品和解决方案。
相关·内容
我试着在读完csv和写了几个转换之后，尝试利用写到pubsbub，但是，一旦我到了需要开始使用pubsub的地步，我就无法导入。以下是我要介绍的内容：
from __future__ import absolute_import
import apache_beam as beam
from datetime import datetime
from beam_utils.sources import CsvFileSource
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from
浏览 0提问于2018-08-25得票数 0
1回答
在流式处理过程中，我尝试将json文件发布到pubsub，并使用cloud Dataflow写入数据存储。
from __future__ import absolute_import
import apache_beam as beam
import json
import logging
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from google.cloud.p
浏览 0提问于2018-10-12得票数 1
我正在尝试通过数据流模板"Pub/Sub Avro to Bigquery“将数据从Pub/Sub流式传输到Bigquery。Pub/Sub中的数据是AVRO格式的，来自Kafka主题。我从模式注册表中获得的相应模式文件。它看起来是这样的： {"type":"record","name":"KsqlDataSourceSchema","namespace":"io.confluent.ksql.avro_schemas","fields":[{"name":
浏览 32提问于2021-01-22得票数 0
我有一个问题，我正在使用python创建数据流模板，该模板在启动新的数据流作业时需要接受3个用户定义的参数。 问题出现在beam.io.gcp.pubsub.WriteToPubSub()中，我试图从ValueProvider提供主题名称，根据谷歌文档，在创建模板时需要使用该名称： https://cloud.google.com/dataflow/docs/guides/templates/creating-templates 源beam.io.ReadFromPubSub()成功地接受了预订值的值提供程序，转换beam.io.gcp.bigquery.WriteToBigQuery()也
浏览 20提问于2020-05-02得票数 1
我的Python包有以下结构，其中beam.py是Dataflow的入口点脚本：
package_name\
__init__.py
tasks\
__init__.py
package_sum.py
utils\
__init__.py
beam.py
.gitignore
requirements.txt
setup.py
Dockerfile
beam.py
import argparse
import apache_beam as beam
from apache_beam.option
浏览 8提问于2021-12-13得票数 0
1回答
我已经用Apache光束写了一个数据流管道，让你对代码有一个基本的了解： Job= (
p
|"cretae">>beam.Create(["message"])
|"job 1" >> beam.ParDo(dofn1())
|"job 2" >> beam.ParDo(dofn2())
|"job 3" >> beam.ParDo(dofn3())
) 目前，我通过创建一个随机消息来触发数据流管道，消息的内容无关紧
浏览 24提问于2021-09-03得票数 0
回答已采纳
1回答
我很难创建一个将发布/子源连接到大型查询接收器的dataflowRunner作业，方法是插入以下两个：
apache_beam.io.gcp.pubsub.PubSubSource
apache_beam.io.gcp.bigquery.BigQuerySink
在github上的beam/sdks/python/apache_beam/examples/streaming_wordcount.py ()示例中分别插入第59行和第74行。删除第61-70行并指定正确的pub/sub和bigquery参数之后，脚本在不构建管道的情况下运行时不会出错。
sidenote:脚本提到流管道支持在Pyt
浏览 0提问于2017-06-29得票数 0
回答已采纳
1回答
我正在尝试网站上的以下示例。
import java.io.IOException;
import org.apache.beam.examples.common.WriteOneFilePerWindow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apac
浏览 12提问于2022-02-03得票数 0
是否有将主题名称作为输入传递的选项？这是一个简单的例子。
class MyOptions(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument(
'--input-topic',
default='projects/project/topics/test'
)
if __name__ == '__main__':
logging.getLo
浏览 3提问于2022-10-03得票数 0
我有一个在pubSub中创建主题并将消息发布到该主题的程序。我还有一个自动的数据流作业(使用模板)，它将这些消息保存到我的BigQuery表中。现在我打算用python管道替换基于模板的作业，我的需求是从PubSub读取数据，应用转换，并将数据保存到BigQuery/发布到另一个PubSub主题。我开始用python编写脚本，并做了很多尝试和错误来实现它，但令我沮丧的是，我无法实现它。代码如下所示：
import apache_beam as beam
from apache_beam.io import WriteToText
TOPIC_PATH = "projects/test-
浏览 3提问于2019-09-16得票数 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
我已经设置了一个用Python编写的自定义DataFlow作业，它只从一个桶复制一个图像(由一个pubsub消息触发)，然后发送一个HTTP。
它看起来像是：
messages = pipeline | 'On Pubsub Message' >> ReadFromPubSub(topic=config['pubsub_image_topic'])
new_images = messages | 'Parse Pubsub Message' >> apache_beam.Map(parse_pubsub)
浏览 7提问于2022-04-15得票数 0
1回答
经过一天的正常工作，从发布/订阅流式传输数据，扁平数据并将行写入BigQuery；数据流管道开始报告如下错误： Processing stuck in step s01 for at least 05m00s without outputting or completing in state process
at sun.misc.Unsafe.park(Native Method)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.
浏览 9提问于2019-04-22得票数 1
1回答
Apache束流管道和毒丸
、、、、
我正在使用GCP数据流运行Apache Beam管道，并从worker那里获得了以下错误：
Error message from worker: java.lang.RuntimeException: java.lang.IllegalStateException: Got poison pill or timeout but stream is not done
两分钟内就有了。
我正在使用管道将消息从PubSub写到BigQuery。在管道中，当将PubSub消息转换为TableRow时，我使用的是FailsafeElement<PubsubMessage, String>，并
浏览 8提问于2022-09-15得票数 0
2回答
我试图从Google和Google存储中读取值，并根据计数条件将这些值放入大查询中，也就是说，如果存在值，那么它不应该插入值，否则它可以插入值。
我的代码如下所示：
p_bq = beam.Pipeline(options=pipeline_options1)
logging.info('Start')
"""Pipeline starts. Create creates a PCollection from what we read from Cloud storage"""
test = p_bq | beam.Creat
浏览 2提问于2018-10-18得票数 1
1回答
如何运行测试？
、、
我正在尝试运行游戏统计、示例、管道和集成测试( )，但我不确定设置本地环境的正确方式是什么。
我的主要目标是学习如何使用TestDataflowRunner，以便为我编写的现有管道实现集成测试。
UPDATE我已经编写了一个基本数据流，它从PubSub读取一条消息并将其写入另一个主题。我有一个使用TestDirectRunner通过的集成测试，但是在尝试使用TestDataflowRunner时会出现错误
pipeline.py
from __future__ import absolute_import
import argparse
import logging
import apa
浏览 2提问于2021-03-18得票数 3
回答已采纳
我正在尝试在GCP上的datalab实例中安装apache beam，但遇到以下问题，我真的很喜欢解决这个问题的方法，目前我对pip或datalab都不是很精通，错误在底部：
Collecting apache-beam[gcp]
Requirement already satisfied: dill<=0.2.8.2,>=0.2.6 in /usr/local/envs/py2env/lib/python2.7/site-packages (from apache-beam[gcp]) (0.2.6)
Requirement already satisfied:
浏览 0提问于2018-08-19得票数 1
我在Google环境中为AppEngine定义了几个不同的云数据流作业。我已经在一个requirements.txt文件中定义了我的需求，包括了我的setup.py文件，而且一切都很正常。我最后一次部署是2018年5月3日。纵观日志，我发现2018年5月22日，我的一份工作开始失败。作业失败，导致堆栈跟踪导致错误导入，如下所示。
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py", line 582, i
浏览 1提问于2018-06-06得票数 1
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
1回答
我正在尝试使用Apache和Dataflow从kafka主题中读取数据，将数据打印到控制台，最后将它们写到公共子主题中。但是它似乎被困在了ReadFromKafka函数中。卡夫卡主题中有很多数据，但是当它运行时，在这个过程中什么也不会发生。
import apache_beam as beam
import argparse
from apache_beam.io.kafka import ReadFromKafka
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.option
浏览 1提问于2021-11-30得票数 4
1回答
我想检查一下云存储中的新文件，我使用的是云发布/订阅。在做了一些分析之后，我想将它保存到另一个云存储中。从这个云存储中，我将使用另一个发布子和数据流提供的模板将文件发送到BigQuery。
当我运行我的代码时，我得到了以下错误：
Traceback (most recent call last):
File "SentAnal.py", line 71, in <module>
"Splitting_Elements_of_Text" >> beam.ParDo(Split()) |
File "C:\Python
浏览 0提问于2019-03-07得票数 2
回答已采纳
我已经创建了一个从PubSub主题解析单个JSON的光束数据流管道： {
"data": "test data",
"options": {
"test options": "test",
"test_units": {
"test": {
"test1": "test1",
"test2":
浏览 18提问于2019-04-18得票数 1
回答已采纳
我目前正在使用Dataflow来处理/按摩来自PubSub的XML文本字符串。我成功地使用DirectRunner为我的-runner标志运行了一个数据流作业。但是，我遇到了一个问题，试图使用与DataflowRunner完全相同的数据流作业来创建数据流资源。
从错误日志(在使用DataflowRunner时)来看，我创建的Dataflow模板似乎不识别：
import xml.etree.ElementTree as ET
每当我在管道中引用ET时，我都会收到"NameError: name‘ET“在运行’generatedPTrans-419‘时没有定义。奇怪的是，我的数据流工作在
浏览 1提问于2020-07-07得票数 0
正如标题所说， 当WriteToBigtable()失败时，是否可以在pubsub中写入死信主题？我知道使用Bigquery是可能的，但还没有在BigTable上看到任何东西。 我使用的是Apache Beam python库。
浏览 11提问于2021-01-19得票数 1
回答已采纳
1回答
我有一个GCS，在这里我每分钟都能得到文件，我使用ApacheBeampythonsdk.i创建了一个用于输入gcs桶和输出gcs bucket.my数据流的发布/子主题创建了一个流数据流，但是我的输出没有存储在输出bucket.this中，这是我的下面的代码。
from __future__ import absolute_import
import os
import logging
import argparse
from google.cloud import language
from google.cloud.language import
浏览 2提问于2019-03-07得票数 3
回答已采纳
我有一个波束管道，从PubSub读取并在应用一些转换后写到BigQuery。管道与NPE一致失败。我使用的是beam 0.6.0版本。知道我可能做错了什么吗？我正在尝试使用DirectRunner运行管道。
java.lang.NullPointerException
at org.apache.beam.sdk.io.PubsubUnboundedSource$PubsubReader.ackBatch(PubsubUnboundedSource.java:640)
at org.apache.beam.sdk.io.PubsubUnboundedSource$PubsubCheckpoin
浏览 4提问于2017-03-23得票数 0
回答已采纳
class load_side_input(beam.DoFn):
def process(self,pubsub_message):
message = pubsub_message.decode("utf8")
output:typing.Dict={}
for key in message.keys():
output[key] = self.tag_model[key]
return [output]
side_input = (p
浏览 14提问于2022-11-19得票数 1
我正在尝试使用光束编程框架(Python SDK)流式传输来自发布/订阅主题的消息，并将它们写出到控制台。 这是我的代码(使用apache-beam==2.27.0)： import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
TOPIC_PATH = "projects/<project-id>/topics/<topic-id>"
def run(pubsub_topic):
options = Pipeline
浏览 24提问于2021-01-21得票数 0
我试着从GCS的发布/订阅主题中读取数据，
options = PipelineOptions(pipeline_args)
options.view_as(SetupOptions).save_main_session = True
options.view_as(StandardOptions).streaming = True
with beam.Pipeline(options=options) as p:
events = p | "ReadPubSub" >> beam.io.ReadStringsFro
浏览 0提问于2019-08-13得票数 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
问题所在
我正在编写一个GCP云函数，它从pubsub消息中获取输入id，处理，并将表输出到BigQuery。
代码如下：
from __future__ import absolute_import
import base64
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from scrapinghub import ScrapinghubClient
import os
def processing_data_function():
# d
浏览 12提问于2020-09-22得票数 1
2回答
我正在尝试用Dataflow创建一个流管道，它从PubSub主题中读取消息，最终将它们写入BigQuery表中。我不想使用任何数据流模板。
目前，我只想在从Google实例执行的Python3脚本中创建一个管道，以执行来自Pubsub的每条消息的加载和转换过程(解析其中包含的记录并添加一个新字段)，最终将结果写入BigQuery表。
简化后，我的代码将是：
#!/usr/bin/env python
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import pubsub_v
浏览 0提问于2019-11-14得票数 3
回答已采纳
当我声明一个具有两个源(1gcs和1pubsub)的管道时，我得到了一个错误，但仅限于波束DirectRunner。使用Google数据流运行器，它工作得很好。我的管道有"Streaming option = True“ gcsEventsColl = p | "Read from GCS" >> beam.io.ReadFromText("gs://sample_events_for_beam/*.log") \
| 'convert to dict' >> beam.Map
浏览 16提问于2021-06-25得票数 0
1回答
我目前正在尝试创建基于ApacheBeamSDKv2.1.0的数据流模板，类似于
这是我的主修课
public static void main(String[] args) {
// Initialize options
DispatcherOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(DispatcherOptions.class);
// Create pipeline
Pipeline pipeline = Pipeline.create
浏览 10提问于2017-09-22得票数 0
回答已采纳
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回答
我是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
回答已采纳
1回答
当我试图导入apache时，我会得到以下错误。
>>> import apache_beam
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/toor/pfff/local/lib/python2.7/site-packages/apache_beam/__init__.py", line 78, in <module>
from apache_beam import io
浏览 1提问于2017-09-19得票数 12
回答已采纳
我正在试用shell中的BeamSQL，并想测试无界源代码在可用性和性能方面是如何工作的。通过阅读here上的文档，我创建了一个外部表，如下所示- CREATE EXTERNAL TABLE pubsub_table (event_timestamp TIMESTAMP, attributes MAP<VARCHAR, VARCHAR>, payload ROW<id INTEGER, location VARCHAR>)
TYPE pubsub
LOCATION 'projects/test-project/topics/test-topic'; 现
浏览 23提问于2019-05-16得票数 1
回答已采纳
我们目前正在向云PubSub发布有关来自谷歌云存储的SFTP到达的事件，我们可以很容易地实时处理这些事件，但当我们尝试从谷歌云存储读取ZIP文件时，我们会遇到一个问题。apache beam的Python端使得它与API中的这个类非常无缝，例如：
import apache_beam as beam
gcs = beam.io.gcsio.GcsIO()
为了能够相对轻松地从Google Cloud Storage中提取文件，我想知道在Java版本中是否有与此python版本等效的版本。很多文档都指向使用MCR中的，我们之前已经做过了，只是感觉比python版本笨重得多，并希望得到一些关于如
浏览 0提问于2017-10-02得票数 0
2回答
如果有人能帮助我理解下面问题中的代码示例，我将不胜感激。我现在正尝试用apache beam 2.13.0和python3.7.3来实现类似的东西。
我知道network sockets是不可序列化的，因为它不是在序列化之后既不能返回字符串也不能返回元组的对象。
我不明白的是，为什么需要在super class中调用__init__？
class PublishFn(beam.DoFn):
def __init__(self, topic_path):
self.topic_path = topic_path
super(self.__class__,
浏览 0提问于2019-07-13得票数 0
回答已采纳
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
回答已采纳
我正在使用 with Apache构建一个简单的管道，该管道将从查询中获取数据，并使用以下代码将数据发布到pub/sub：
package main
import (
"context"
"flag"
"github.com/apache/beam/sdks/go/pkg/beam"
"github.com/apache/beam/sdks/go/pkg/beam/io/pubsubio"
"github.com/apache/beam/sdks/go/pkg/beam/log&#
浏览 3提问于2021-10-20得票数 7
当将固定数目的字符串(用于测试的80万1KB )放入PubSub主题并在Dataflow中运行以下Apache (2.1.0)作业时，完全按照预期保持一次语义。
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transfo
浏览 0提问于2017-11-17得票数 0
回答已采纳
1回答
我正在学习Apache中的窗口&触发概念，目的是：
读取无界源(PubSub)
每5秒将传入消息写入本地主机磁盘，每5秒固定窗口间隔。
问题:没有输出被写入本地主机磁盘(管道确实创建了一个波束团队文件夹，并在其中写入了一些文件，但是目标目标中没有output.csv，每5秒编写一次)。
运行apache-beam=2.9.0，Python 2.7.10
尝试了两种方法：DirectRunner，和DataFlowRunner (以GCS桶为目的地)
下面是代码(谢谢您的建议)：
p = beam.Pipeline(runner=None, options
浏览 0提问于2019-01-02得票数 2
我需要通过apache- need将数据从pub/sub写入Bigquery。代码如下：
import argparse
import base64
import logging
import json
from datetime import datetime
from ast import literal_eval
from google.cloud import bigquery
import apache_beam as beam
class DataIngestion(beam.DoFn):
@classmethod
def parse_method(cls,
浏览 0提问于2019-12-05得票数 0
1回答
尝试在GCP云函数中创建GCP数据流。我已经部署了一个简单的apache函数，它工作得很好，但是当我试图读取文件时会出现路径错误。当我使用参数-runner从本地运行时，与Dataflowrunner一样，相同的脚本运行，有人建议我必须执行pip安装apache-beamgcp。我已经在当地做过了，而且效果很好。如果我试图在GCP中安装它，它会在一段时间后进行会话超时。下面是我的密码。
#import print library
# This script will read all avro files on a path and print them
import logging
imp
浏览 11提问于2022-08-04得票数 1
回答已采纳
我们有一个用Java编写的beam管道，我们运行在GCP数据流上。它非常简单，它将一个SQL查询作为一个PipelineOption，针对BigQuery和返回数据集中的每一行进行SQL查询，构造一条消息并将其放到一个公共主题上。
import com.google.api.services.bigquery.model.TableRow;
import java.util.HashMap;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
浏览 3提问于2022-03-01得票数 0
回答已采纳
试图做一些概念上简单的事情，但却让我的头撞到了墙上。
我尝试在Python语言中创建一个流式数据流作业，它使用来自PubSub主题/订阅的JSON消息，对每个消息执行一些基本操作(在本例中，将温度从C转换为F)，然后将记录发布回不同的主题：
from __future__ import absolute_import
import logging
import argparse
import apache_beam as beam
import apache_beam.transforms.window as window
import json
'''Normal
浏览 3提问于2018-06-29得票数 1
我们运行了一个Google批处理作业，它将数据记录写入一个pubsub管道，并有一个单独的流作业，它使用“pubsub to elasticsearch”流模板从PubSub管道中提取数据，并将更新写入我们的Elasticsearch索引。但是，我们不得不终止写入Elastic的流作业，因为在1次尝试之后，我们遇到了多个“向ES写入错误”。再也不允许“错误”了。
从PubSub读取并写入Elasticsearch的任务是写入不是流索引的索引。它实际上需要在索引中更新相同的文档，可能会多次更新。
这项工作需要写入流索引吗？如果不是，我们如何配置数据流作业，使其实际上放慢速度，而不是压倒弹性？
我
浏览 16提问于2022-09-15得票数 0
1回答
我有一个使用DataflowRunner部署的apache streaming流水线。我的管道代码看起来像这样：
with beam.Pipeline(options=pipeline_options) as p:
(p | "Read input from PubSub" >>
beam.io.ReadFromPubSub(subscription=known_args.subscription)
# ...
然后，我像这样部署管道，python3 main.py --runner=DataflowRunner --streaming
浏览 3提问于2021-02-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
