使用发布/订阅和数据流从单个JSON创建和插入多行到BigQuery
使用发布/订阅和数据流从单个JSON创建和插入多行到BigQuery是一种将数据从源系统传输到Google Cloud的方法。下面是对这个过程的详细解释：
发布/订阅模式是一种消息传递模式，其中消息发布者将消息发送到特定的主题（Topic），而订阅者则从该主题订阅消息。在这种情况下，我们可以将JSON数据作为消息发布到一个主题。
数据流（Dataflow）是Google Cloud提供的一种托管式数据处理服务，用于在大规模数据集上进行ETL（提取、转换和加载）操作。数据流可以接收来自发布/订阅主题的消息，并将其处理后插入到BigQuery中。
BigQuery是Google Cloud提供的一种快速、可扩展且完全托管的云数据仓库。它可以用于存储和分析大规模数据集，并提供了强大的查询和分析功能。
使用发布/订阅和数据流从单个JSON创建和插入多行到BigQuery的步骤如下：
- 创建一个发布/订阅主题：在Google Cloud控制台中，创建一个主题，用于接收JSON数据。
- 配置数据流作业：使用Google Cloud的数据流服务，创建一个数据流作业。在作业配置中，指定要从发布/订阅主题接收消息，并将其插入到BigQuery中。
- 定义数据转换逻辑：在数据流作业中，您可以定义数据的转换逻辑。这可以包括解析JSON数据、转换数据格式、筛选数据等操作。
- 配置目标表：指定要将数据插入的BigQuery表。您可以创建一个新表或将数据追加到现有表中。
- 启动数据流作业：启动数据流作业后，它将开始从发布/订阅主题接收消息，并将其插入到BigQuery中。
这种方法的优势包括：
- 实时数据处理：使用发布/订阅和数据流，您可以实现实时数据处理，将数据从源系统传输到BigQuery，以便进行实时分析和查询。
- 弹性扩展：数据流是一种完全托管的服务，可以根据数据量的变化自动扩展计算资源，以确保高性能和可靠性。
- 简化的开发和维护：使用数据流和BigQuery，您无需担心基础设施的管理和维护，可以专注于数据处理逻辑的开发和优化。
- 高可靠性和持久性：数据流提供了消息传递的可靠性保证，确保消息不会丢失，并提供了至少一次的传递保证。
- 强大的查询和分析功能：通过将数据插入到BigQuery中，您可以利用其强大的查询和分析功能，对大规模数据集进行复杂的查询和聚合操作。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CMQ：https://cloud.tencent.com/product/cmq
- 腾讯云数据流计算 TDSQL：https://cloud.tencent.com/product/tdsql
- 腾讯云云数据库 TencentDB：https://cloud.tencent.com/product/cdb
- 腾讯云云原生容器服务 TKE：https://cloud.tencent.com/product/tke
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云人工智能 AI Lab：https://cloud.tencent.com/product/ai
- 腾讯云物联网平台 IoT Hub：https://cloud.tencent.com/product/iothub
- 腾讯云移动开发 MSDK：https://cloud.tencent.com/product/msdk
- 腾讯云对象存储 COS：https://cloud.tencent.com/product/cos
- 腾讯云区块链服务 TBaaS：https://cloud.tencent.com/product/tbaas
- 腾讯云元宇宙服务：https://cloud.tencent.com/product/virtual-universe
相关·内容
我在做一个项目，在这个项目中我遇到了一个问题，我想从BigQuery数据集中获取实时数据，所以我努力研究了解web套接字，如pusher和laravel-websocket，我还对它们进行了测试 但是有一件事我不明白，如果我把Laravel SDK和我的BigQuery项目连接起来，我没有发现BigQuery提供给我们的任何事件或方法，因为在我的场景中，大多数数据来自IoT设备，而设备使用bigQueryAPI来提供数据，我也希望这些设备的数据是实时的。 然后，我发现有一种方法可以将BigQuery dataset与firebase连接起来，然后每当数据集中的行进给firebase时，fir
浏览 20提问于2021-07-07得票数 0
我正在使用GCP处理和存储从模拟IoT设备发送的数据。数据(应该)在以下路径中流动： 带有mqtt客户端的模拟IoT设备-> IoT核心设备-> IoT核心注册表->发布/订阅主题->发布/订阅订阅->云数据流发布/订阅订阅BigQuery模板-> BigQuery表 我所在的公司有一些GCP配置(它在VPC上)，这需要我指定一个subnet for DataFlow to even run。当数据流在指定子网后运行时，它不会执行任何操作。它不会从订阅中提取数据，也不会输出任何错误消息。它显然没有意识到数据正被发送到主题和订阅。我还尝试了云数据流发布/订阅
浏览 21提问于2021-10-01得票数 0
回答已采纳
我创建了一个google流作业，从PubSub读取并插入到BigQuery中。我使用JSON将STREAMING_INSERT数据插入到BiqQuery表中。我正面临插入问题，说明请求的大小超过了10 Mb的允许限制。数据流错误如下所示。根据我的理解，每个记录的大小是1-2Mb，根据我的理解，数据流作业将流数据插入到微批处理中，这导致了这一错误。
你能不能给我一些解决办法。
来自worker的错误消息: java.lang.RuntimeException:我们观察到一行大小为24625273字节。BigQuery支持高达10 is的请求大小，这一行太大了。您可以更改重试策略，以取消阻塞该管
浏览 6提问于2022-10-13得票数 0
回答已采纳
我正在使用pubsub中的“”特性，通过数据流将一个常规的JSON传递给bigquery。
但是，它工作了一秒钟，这意味着一些条目可以正确地传递到bigquery。但是现在我在数据流日志上发现了错误
java.lang.RuntimeException: java.io.IOException:插入失败：[{“错误”：{“debugInfo”：“”、“位置”：“_comments”、“消息”：“没有这样的字段”、“原因”：“无效”}，“索引”：0}] org.apache.beam.sdk.io.gcp.bigquery.StreamingWriteFn.flushRows(Streamin
浏览 1提问于2018-03-27得票数 1
回答已采纳
1回答
我想使用Cloud Dataflow、PubSub和Bigquery将tableRow写入发布订阅消息，然后将它们写入Bigquery。我希望表名、项目id和数据集id是动态的。
我在互联网上看到了下面的代码，我不能理解如何传递数据行参数。
public void PubSub(String projectId , String datasetId,String tableId,String topicId)
PipelineOptions options = PipelineOptionsFactory.create();
DataflowPipelineOpt
浏览 0提问于2018-07-15得票数 0
我能够创建一个数据流管道，它从发布/订阅读取数据，并在处理后以流模式写入大型查询。
现在，我希望以批处理模式运行我的流水线，而不是流模式，以降低成本。
目前，我的流水线正在使用动态目的地的bigquery进行流式插入。我想知道是否有一种方法可以对动态目的地执行批量插入操作。
下面是
public class StarterPipeline {
public interface StarterPipelineOption extends PipelineOptions {
/**
* Set this required option to specify where
浏览 3提问于2018-06-14得票数 2
2回答
我正在使用Cloud Dataflow streaming管道将从发布/订阅接收的事件插入到BigQuery数据集中。我需要几个，以保持每个工作简单，易于维护。
我担心的是全球成本。数据量不是很大。在一天中的几个时段，没有任何数据(发布/订阅上的任何消息)。
我希望在收到新消息之前，数据流的比例为0 worker。但是看起来最小工人数是1。
因此，每项工作的最低价格为一天: 24 vCPU小时…所以每个月至少50美元/份工作。(月度使用不打折)
我计划每天通过api运行和耗尽我的作业几次，以避免一个全职工作人员。但是对于像DataFlow这样的托管服务来说，这似乎不是正确的形式。
我是不是漏掉
浏览 0提问于2018-09-12得票数 4
我希望将数据流到BigQuery中，并且我正在考虑使用PubSub + Cloud函数，因为不需要进行转换(至少目前是这样)，而且使用云数据流似乎有点过分，只需将行插入到表中即可。我是对的？
数据是使用Python脚本从GCP流到PubSub中的，其格式如下：
{'SEGMENT':'datetime':'2020-12-05 11:25:05.64684','values':(2568.025,2567.03)}
BigQuery模式是datetime:timestamp, value_A: float, value_B: fl
浏览 10提问于2020-12-05得票数 1
回答已采纳
1回答
为什么使用发布/订阅？Usecase:有一个http触发的"Cloud function“，在这个函数上发送一些数据。此函数在处理数据后，将数据发布到发布/订阅主题。 然后还有另一个Cloud函数，它是基于发布到该发布/订阅主题而触发的。因此，这个云函数从发布/订阅中获取已发布的数据，并将其插入到BigQuery表中。 所以在这个使用pub sub的用例中，为什么我们不能只有一个云函数，它从http中获取数据，并将其插入到BigQuery中。在这里选择pub/sub的设计思路是什么？ 另外，为什么通常使用发布/订阅架构？
浏览 7提问于2021-04-09得票数 1
回答已采纳
目前，我正在使用DataFlow进行流处理，将上传的blobs从GCS迁移到BigQuery。但是，我发现有几条指向同一个objectId的pub/sub消息导致了BigQuery端的复制。
我知道GCS通知保证至少一次传递，因此对于给定的blob文件，我可能会有重复的pub/sub消息。
我阅读了Python文档，它可以使用id_label作为唯一的记录标识符。为了根据blob文件名去重复，我尝试在attributes.objectId args上插入id_label。尽管如此，它似乎不起作用(我仍然收到重复的消息)：
| "Read from pub/sub topic"
浏览 11提问于2022-04-28得票数 0
我正在尝试编写一个使用GCP服务的摄取应用程序。每天可能有大约1TB的数据以流的方式到来(即每小时100 TB，甚至在特定时间一次)。 我正在尝试设计一个摄取应用程序，我首先认为在cron作业中编写一个简单的Python脚本来顺序读取文件(甚至在两个三线程中)，然后将它们作为消息发布到pub/sub是一个好主意。此外，我需要有一个数据流作业运行，始终从发布/订阅读取数据，并将它们保存到BigQuery。 但是我真的想知道在这里我是否需要发布/订阅，我知道数据流可以非常灵活，我想知道我是否可以将1TB的数据作为批处理作业直接从GCS摄取到BigQuery，或者更好的做法是通过我上面提到的流式作
浏览 4提问于2019-02-15得票数 1
1回答
我将以下json插入到预先创建的BigQuery表中。
{
"Member_ID": 881230,
"First_Name": "Dave2",
"Last_Name": "Manin2",
"Gender": "M",
"Age": 53,
"Height": "5,2",
"Weight": 145,
"Hours_Sleep": 4,
"Calories
浏览 3提问于2018-11-14得票数 3
我需要用Python语言从PubSub写BigQuery。我测试了一些异步用户代码，它工作得很好。但这需要持续运行，我不是100%确定在哪里安排。我一直在使用Cloud Composer (Airflow)，但它看起来不太适合，而且看起来Dataflow是GCP推荐的？对吗？
或者，有没有办法从Cloud Composer中可靠地运行它？我想我可以运行它一次，但我想确保它再次运行，以防它由于某种原因而失败。
浏览 3提问于2018-11-25得票数 1
1回答
我们的IoT设备收集样本数据，样本数据无限期地停留在设备上。而且，我们的IoT设备不会收集大量的样本，最多每周采集十几个样本。
我正在使用AWS IoT，我已经成功地将我的设备连接到IoT核心，并且我可以通过IoT协议发布/订阅消息。从这里开始，我有一条规则将有效负载转发到我的HTTP端点。
在使用AWS IoT之前，我通过将json有效负载从我的设备发送到我的http端点来验证概念。此操作将新记录插入到我的RDS数据库中，并将UUID返回给设备。然后，该设备用相应的示例存储UUID。通过这种方式，我们能够将设备上的示例与云服务中的样本进行匹配。
然而，现在我已经过渡到使用MQTT，我意识到我
浏览 0提问于2020-08-31得票数 0
回答已采纳
1回答
我关注的是通过GCD发布/订阅的物联网实时流处理，云数据流和通过BigQuery.I执行分析。我正在寻求如何实现这一点的帮助。
浏览 2提问于2017-06-15得票数 1
我一直在尝试在Airflow中使用PubSubPullSensor来拉取JSON消息并将其摄取到bigquery中，当消息大小超过阈值时，传感器无法通过XCOM拉取消息并推送。我知道XCOM有一个最大大小限制，但是我们如何克服这种情况，对于这种情况有什么变通办法吗？
浏览 3提问于2020-05-12得票数 0
我正在浏览谷歌的Firestore to BigQuery pipeline tutorual，我已经到了第10步，在那里我应该设置从我的主题导出到保存在云存储上的avro文件。 但是，当我尝试运行作业时，在执行完教程中提到的操作后，我得到一个错误，告诉我我的项目没有足够的配额来执行工作流。在消息的配额摘要中，我注意到它显示为1230/818 disk GB。这是否意味着该作业需要1230 GB的磁盘空间？目前，Firestore中只有100个文档。在我看来这是错的？ ? 我所有的云存储桶都是空的： ? 但是，当我查看我在page 9上设置的第一个导出作业(从Pubsub主题到Big
浏览 26提问于2019-05-23得票数 0
回答已采纳
我目前在GCP中有一个管道，它使用云函数摄取数据，将其存入PubSub，在数据流中进行处理，最后将其插入到BigQuery中。我正在尝试计算平均消息从云函数摄取到BigQuery需要多长时间(包括插入时间)。
我了解Stackdriver指标/仪表板以及GCP的API仪表板。我很想简单地将每个服务的所有延迟值相加，但我不确定它是否涵盖了我无法控制的延迟值(例如，插入到BQ，从数据流推送到pubsub，等等)
哪些可用指标可以帮助我计算整体流水线延迟？如果管道的一部分无法计算，是否有方法估计它或使用外部工具来计算它？
浏览 2提问于2019-11-21得票数 0
我想使用pubsub将数据插入到bigquery表中。数据已经处理过了，所以我不需要数据流。我该怎么做呢？提前感谢
浏览 0提问于2018-07-03得票数 2
我对Flink相当陌生，在技术用例方面需要一些帮助。
当前场景：
我有一个在GKE上运行的flink应用程序，并使用自定义接收器将记录(从Kafka源)写入BigQuery。我可以将记录写入BigQuery，而不会出现任何问题。目前，记录被一个接一个地写入接收器中，因此每个kafka消息都会得到自己对BigQuery的插入api调用，这并不理想，因为我们需要执行批量插入，并且单独插入每个记录将非常昂贵。为此，我使用了bigquery-storagewrite。
新要求：
在将记录写入BigQuery之前，理想情况下，我希望在将它们写入接收器之前根据大小/时间对记录进行缓冲。
我不知道如何在Fl
浏览 5提问于2022-10-18得票数 2
回答已采纳
我目前正在创建一个BigQuery表，该表具有带有Terraform的。我想通过BigQuery PubSub订阅插入数据。PubSub订阅的主题模式是用AVRO编写的。我在跟踪文档来映射字段，但是由于JSON仍然是BigQuery的实验对象，所以我找不到相应的docu。当我试图应用我的代码时，我得到：
Error: Error creating Subscription: googleapi: Error 400: Incompatible schema type for field data: STRING vs. JSON Details: [ {
"@type
浏览 16提问于2022-10-14得票数 0
回答已采纳
我们尝试使用数据流将数据插入到bigquery (流)中。有没有办法可以检查插入到Bigquery中的记录的数量？我们需要这些数据来进行对账。
浏览 19提问于2019-03-26得票数 0
我需要将100万条(甚至更多)记录从Server表插入到BigQuery表中，该表作为“链接服务器”存在于Server中，通过中的远程处理守护进程作为“链接服务器”存在。
此外，源表可能没有行数、Id等数的列。
现在，我可以使用此驱动程序将每秒1记录插入到BigQuery中，使用以下查询：
INSERT INTO [GBQ].[CDataGoogleBigQuery].[GoogleBigQuery].[natality]
SELECT *
FROM [natality].[dbo].[natality]
GO
但是对于这么多的记录，比如一百万甚至更多，这是一个非常低的性能
浏览 6提问于2015-10-29得票数 0
我有一个每秒发布大约1条消息的pubsub主题。消息大小约为1kb。我需要将这些数据实时放到cloudsql和bigquery中。数据正在以稳定的速度到来，关键是它们都不会丢失或延迟。将它们多次写入destination不是问题。数据库中所有数据的大小约为1 1GB。
使用主题触发的google云功能与google dataflow相比解决此问题的优缺点是什么？
浏览 26提问于2020-02-27得票数 0
回答已采纳
使用云函数的推荐方法之一是使用发布/订阅推送订阅来调用它们，以将发布/订阅消息写入BigQuery。我有非常小的消息，但我有很多消息。
我为写到GBQ的60.000.000条消息支付了大约80美元。我的GCF配置为128MB，超时1秒。
源代码：
exports['write-to-gbq'] = async (event, context) => {
const message = JSON.parse(Buffer.from(event.data, 'base64').toString())
const dataset = bigquery.d
浏览 0提问于2021-10-17得票数 0
我在bigquery中有一个巨大的数据集，有5000万行和57列。我想做很多过滤/转换/清理，而不是使用sql。我尝试使用dask/panda/python将数据加载到本地mac的dask数据帧中，进行转换，然后将数据推送回bigquery，以便其他总线可以使用它。将数据推送回bigquery需要超过3个小时。有没有其他方式，或者谷歌云服务，我可以利用？
浏览 2提问于2019-03-30得票数 0
我在一家初创公司工作，在那里我们可以销售某种IoT设备。这些设备将连接到我们托管在谷歌云中的服务器，并将每1秒发送一次数据，我的服务器将其作为时间序列存储在数据库中。假设我们连接了1000台设备，所有设备都在每秒发送数据，是否适合使用google bigquery每秒将这些数据插入表中，将这些数据插入到设备所有者的相应表中？
由于我的数据是时间序列的形式，我正在考虑对每个用户(我的设备的所有者)使用分区表，但由于官方文档中列出的限制和配额，我担心每秒大量插入会达到限制(这并不是说我将根据手机应用程序上的用户需求来查询数据)。
如果它不合适，那么什么适合我的用例呢？
编辑:我主要关心的是每秒插入
浏览 36提问于2018-05-31得票数 0
我发现bigquery中的预定查询不具备在依赖表运行出错时发出通知的功能。
例如，table A的创建是基于table B的，如果table A今天运行该错误，我需要bigquery通知我table A今天有问题。还有什么工具或软件可以帮助我解决这个问题？
浏览 17提问于2021-10-18得票数 0
1回答
我有大约1500个作业要用Dataflow来实现。这些工作将每天安排。我们可以在作业中使用Bigquery客户端库使用大量的DML语句。列出我对Bigquery配额和限制的关注。
参考资料：https://cloud.google.com/bigquery/quotas
请确认我们是否需要在以下任何一个场景中考虑Bigquery的每日使用限制。
如果我们使用BigqueryIO.write()实现数据插入
如果在Dataflow作业中使用Bigquery客户端库使用DML语句(更新/删除)
请建议一下。
浏览 1提问于2017-11-28得票数 0
回答已采纳
我使用谷歌提供的模板从PubSub转到BigQuery，没有进行任何定制。我尝试将多个条目(行)放入到队列的单个json有效负载中，然后让DataFlow模板将所有条目(行)插入到BigQuery表中。我已经尝试提供一个换行符分隔的json有效负载，就像通过控制台将数据加载到BigQuery中时所需要的一样。但是，我只能获得要插入到表中的第一个条目。
默认的DataFlow模板是否只接受一个条目(行)？
浏览 1提问于2018-03-17得票数 0
我在Google Dataflow中运行了一个Apache beam流水线。它从Kafka中读取数据并将其流式插入到Bigquery。
但在bigquery流插入步骤中，它抛出了大量警告-
java.lang.RuntimeException: ManagedChannel allocation site
at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init> (ManagedChannelOrphanWrapper.java:93)
at io.grpc.internal
浏览 7提问于2021-06-01得票数 5
我想使用将来自主题的PubSub消息数据插入到BigQuery表中。一切都很好，但是在BigQuery表中，我可以看到诸如“߈���”这样的不可读字符串。这是我的管道：
p.apply(PubsubIO.Read.named("ReadFromPubsub").topic("projects/project-name/topics/topic-name"))
.apply(ParDo.named("Transformation").of(new StringToRowConverter()))
.apply(BigQueryIO.Write.na
浏览 1提问于2015-09-17得票数 8
回答已采纳
我想读取Google PubSub主题上的传入数据，处理数据并将其转换为统一的数据结构，然后将其插入到Google BigQuery中的数据集中。据我所知，可以使用某种流数据的管道。然而，我很难找到任何好的和简洁的例子来实现这一点。
我的项目是用Scala编写的，所以我更喜欢用该语言编写的示例。否则，Java中的一些简洁的东西也会起作用。
谢谢!
浏览 2提问于2018-05-03得票数 0
回答已采纳
我正在使用BigQuery连接器在WSO2中插入BigQuery云中的多个记录。
我遵循了这个并成功地插入了一条记录。
First我试图通过使用逗号分隔它们来传递多个记录。
但是，在这种情况下，只插入第一条记录并跳过其他记录。
请注意，在这种情况下不返回错误。
我的Json消息有多条记录：
{"insertId":"101","json":{"NAME":"Vishal_101","ADDRESS":"UK","ID":"vbordia"}},{
浏览 3提问于2019-03-18得票数 0
有人在数据流中使用过DynamicDestination，他有一个简单且描述过的示例。我厌倦了在git (<code>C0</code>)中看到远程传送的例子，它伤害了我作为一个新手在阿帕奇光束。顺便说一下，我需要做的是从Pubsub读取消息，并通过数据流作业写入BigQuery dataset中的不同目的地(表)。我有一个自定义项目，它非常适合Bigquery表，但Pubsub主题将包含来自同一数据集的多个目的地。此外，该消息是JSON格式的，并且包含一个带有目标表名称的字段。
这是我最具代表性的代码
<code>A1</code>
有什么建议
浏览 15提问于2020-03-18得票数 2
基本上我已经创建了一个云函数(写了一个Node.js代码)，它将在云发布主题的消息上触发，并将数据加载到Bigquery表中。
topic中的消息被云函数读取后被删除。我知道订阅者在内部发送确认和结果，消息从主题中删除。
我想控制发送给发布者的确认。怎么才能实现，没有找到相关的文档。
浏览 2提问于2017-10-18得票数 12
当我把这件事作为我的论文项目时，我在整个数据工程领域真的是个新手，所以请容忍我。
我目前正在为一个电池存储系统开发一个大数据平台，它已经有了每15秒收集一次数据的CloudSQL服务(所以它已经在GCP环境中了)。我的工作是复制它们并在每次A数据传入时传输到BIGQUERY (包括使用Dataprep准备数据)，然后将其实现为机器学习。
我已经找到了几种方法，其中一种是使用数据流，尝试过一次，但它是手动完成的。使用jdbc to bigquery作业。为了满足我的需求(定期运行作业)，我被推荐使用Cloud Composer。
另一方面，我得到了另一个使用PubSub的源，它将作业触发到数据流
浏览 6提问于2020-11-20得票数 0
1回答
我有一个云函数，它将消息发布到PubSub，并触发云运行来执行归档文件过程。当有大文件时，我的cloud Run python代码需要一些时间来处理数据，它看起来像是PubSub在20秒(默认确认截止时间)后重试消息，这将从我的Cloud run触发另一个实例。我已经将确认截止时间增加到600秒，并重新部署了所有内容，但它在20秒后仍在重试消息。我遗漏了什么吗？ 云函数发布消息代码： # Publishes a message
try:
publish_future = publisher.publish(topic_path, data=message_bytes)
浏览 17提问于2021-07-02得票数 2
回答已采纳
1回答
你能帮助我提供如何在biGQuery中加载我的谷歌分析数据到Redshift的方法吗？云函数能做到这一点吗？或者，我如何从python端触发它，而不是使用bigquery脚本来调用BigQuery端？
浏览 0提问于2019-11-28得票数 1
如果应用程序使用Java将事件作为行插入到BigQuery表中，那么我希望避免意外地多次插入相同的事件/行(因为错误、重试等等)。
每个事件，我作为一行插入，有一个唯一的UUID，但我没有流偏移量。(它不能保证事件在我的应用程序中以固定的顺序到达，因为它使用的是带有多个分区的Kafka主题。)
使用，我能够使用以下函数的id参数依赖于。
com.google.cloud.bigquery.InsertAllRequest.RowToInsert.of(String id, Map<String, ?> content)
新的 (在提交模式下)确实有一个精确的一次保证，但前提是你可以
浏览 10提问于2022-02-06得票数 0
目前，我有一个系统，希望通过Google函数将该系统中的数据发送到BigQuery (使用Python)。我的想法是分两个简单的步骤：
系统调用Google函数并向该函数发送JSON数据
Google功能将数据解压缩并存储在Google包的Google BigQuery表中。
我认为这是个好主意，直到我意识到谷歌BigQuery在添加数据方面存在利率限制(例如，参见 )。
因为我想使用这个系统每天触发超过一千次，我想我需要一个更好的设置。
你建议我怎么做？
我已经想到了两件事：
使用添加数据，但我不确定该数据是否具有相同的速率限制，以及是否可以轻松地将该数据传输到BigQu
浏览 5提问于2022-06-28得票数 0
回答已采纳
3回答
只有当数据流作业执行成功完成时，我才想触发云函数。
如果数据流作业失败，则不应触发云功能。
我使用数据流模板(jdbc到BigQuery)从Dataflow UI运行数据流作业。
作业执行后，没有任何选项触发任何云函数或其他东西。另外，我不能修改模板代码。触发云功能的方法是什么？
浏览 4提问于2020-11-25得票数 0
回答已采纳
1回答
我试图在Bigquery中对三个表( order、order_item和user )进行去角色化和连接。我已经有了一个管道，用于使用pubsub和dataflow更新这些表。现在，我想要创建一个管道，在提到的三个表中的任何一个发生更新或插入时，更新一个新的、非规范化的表。order和order_item上的这些更新必须是同步的，只有当插入按顺序进行时(以便在创建新订单时显示用户状态)，用户才应该被连接。
到目前为止我想出了两种解决方案。
一种是捕获每个表上的更改，通过数据流读取来自pubsub的消息，然后从非规范化表中丰富最新的相应记录。最后，将旧记录替换为新记录。
另一个是查询ord
浏览 0提问于2019-06-24得票数 1
在上按键执行组之后，我需要：- 1)使Pcollection中的每个元素成为单独的PCollection。2)将单个PCollections中的记录插入到BigQuery表中。基本上，我的目的是在BigQuery表中创建一个动态日期分区。我该怎么做？举个例子会很有帮助。
浏览 2提问于2017-03-03得票数 0
我想了解数据流管道的工作原理。 在我的例子中，我有一些定期发布到云发布/订阅的内容，然后Dataflow将其写入BigQuery。通过的消息量数以千计，所以我的发布者客户端有1000条消息的批处理设置，1mb和10秒的延迟。 问题是，在批处理中发布时，Dataflow SQL是接收批处理中的所有消息并将其一次性写入BigQuery，还是一次写入一条消息？ 有没有一个比另一个更好的好处呢？ 如果需要任何其他细节，请评论。谢谢
浏览 35提问于2021-11-23得票数 0
回答已采纳
我使用Airflow DataflowTemplateOperator()将数据从Mssql迁移到Bigquery，使用JDBC to Bigquery数据流模板。
默认情况下，它会将数据追加到目标Bigquery表中。
我想先截断该表，然后再写入新行。
是否有默认参数可将数据流/ DataflowTemplateOperator设置从追加更改为覆盖？
浏览 2提问于2020-12-02得票数 0
我希望使用Dataflow将事件从PubSub流流到BigQuery。我看到在GCP中有两个模板可以实现这一点:一个是Dataflow从主题中读取消息，另一个是订阅。
在这里使用订阅的好处是什么，而不仅仅是使用主题中的事件？
浏览 0提问于2019-05-24得票数 2
回答已采纳
我们在BigQuery中有一个按日分区的表，它是通过流插入更新的。
说：“当流到分区表时，流缓冲区中的数据对于_PARTITIONTIME伪列有一个空值”。
但是，如果我查询select count(*) from table where _PARTITIONTIME is NULL，它总是返回0，尽管bq show告诉我流缓冲区中有很多行。
这是否意味着流缓冲区中的行根本不存在伪列？在任何情况下，如何只在流缓冲区中查询数据，而不使其成为全表扫描？
提前感谢
浏览 0提问于2017-01-26得票数 8
回答已采纳
我需要一个如何在一个单独的子进程中运行耗时的web服务的理解和代码架构。在一个单独的子进程中运行这样的web服务会运行更长时间，这是不是很好？web服务将接收多达2000个嵌套的json对象的负载，数量可以更少或更多，它将获取数据并插入到数据库中。下面只是代码片段，只是为了给出一个概念。
router.post('/longRunningWebservice', function (req, res) {
//for loop to iterate 2000 json objects
//it has some queries to insert data
浏览 0提问于2016-06-17得票数 0
我想测试一下我的管道。我的管道从BigQuery中提取数据，然后将数据存储到GCS和S3中。虽然这里有一些关于流水线测试的信息，，但它没有包括从BigQuery中提取数据的数据模型。
我找到了下面的例子，但它缺少注释，所以有点难以理解。
有没有好的文档来测试我的流水线？
浏览 15提问于2017-01-25得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
