最佳实践-使用受限服务帐户将无界PCollection写入GCS Bucket
在云计算领域中，最佳实践是使用受限服务帐户将无界PCollection写入GCS（Google Cloud Storage） Bucket。下面是对这个问题的完善且全面的答案：
概念：
- 受限服务帐户（Service Account）是一种用于身份验证和授权的云计算服务帐户。它们是通过密钥对或者其他安全凭据来访问云服务的。
- 无界PCollection（Unbounded PCollection）是指在数据处理流水线中，数据流没有固定的开始和结束点，可以持续地产生新的数据。
分类：
- 受限服务帐户属于身份验证和授权的范畴。
- 无界PCollection属于数据处理流水线的范畴。
优势：
- 使用受限服务帐户可以提供更好的安全性，因为它们具有较小的权限范围，可以限制对资源的访问。
- 无界PCollection可以处理实时数据流，而不需要固定的开始和结束点，适用于需要实时处理大量数据的场景。
应用场景：
- 使用受限服务帐户将无界PCollection写入GCS Bucket适用于需要实时处理数据并将结果存储在云存储中的场景。例如，实时日志分析、实时监控和实时报警等。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云的受限服务帐户：https://cloud.tencent.com/document/product/598/10597
- 腾讯云的GCS Bucket：https://cloud.tencent.com/document/product/436/13309
总结： 最佳实践是使用受限服务帐户将无界PCollection写入GCS Bucket。受限服务帐户提供了更好的安全性，而无界PCollection适用于实时处理数据流的场景。腾讯云提供了相应的受限服务帐户和GCS Bucket产品，可以满足这个需求。
相关·内容
1回答
最近，我开始与Apache和Google的云数据流合作，开发大数据处理管道。我计划利用，开发我的处理管道。
下面是我想要达到的目标的要点
从无界(流)源读取数据，对于GCP，这将是PubSub。将数据转换为KV
从有界源读取数据，对于GCP，我的情况是GCS。将数据转换为KV
由于其中一个源是无界的，我必须做窗口和触发.我选择使用Global，因为我希望将来自无界源的数据连接到有界源，如果它已经存在于我的内部状态管理中。
在公用密钥上连接来自PubSub和GCS的数据。
使用 (StateId和StateSpec)将数据添加到Beam的内部状态。发出添加到BagState的
浏览 0提问于2019-03-19得票数 1
试着让我的问题尽可能宽泛： 当使用TextIO将无界PCollection写入GCS存储桶时，同时使用没有GCS删除访问权限的最小权限原则的服务帐户时，数据流中出现以下错误： Error trying to copy gs://[Temporary beam file] to gs://[JSON We expect]: {"code":403,"errors":[{"domain":"global","message":"[Service Account] does not have storage.o
浏览 14提问于2019-03-04得票数 0
回答已采纳
我正在使用apache beam向gcs编写一个json。但是遇到了以下错误 NotImplementedError: offset: 0, whence: 0, position: 50547, last: 50547 [while running 'Writing new data to gcs/write data gcs/Write/WriteImpl/WriteBundles/WriteBundles'] 不知道为什么会发生此错误。相同的代码如下： class WriteDataGCS(beam.PTransform):
"""
浏览 20提问于2019-09-11得票数 0
回答已采纳
我正在使用数据流中的流式数据从pub-sub中收听数据。然后，我需要上传到存储，处理数据并上传到bigquery。
这是我的代码：
public class BotPipline {
public static void main(String[] args) {
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setRunner(BlockingDataflowPipelineRunner.class);
opti
浏览 1提问于2016-11-02得票数 1
回答已采纳
2回答
我见过许多关于同一主题的问题。但是，我在给GCS写信方面仍然有问题。我从公共图书馆读到了这个话题，并试图将其推广到GCS。我提到过。但是，在最新的束流封装中找不到IOChannelUtils。
PCollection<String> details = pipeline
.apply(PubsubIO.readStrings().fromTopic("/topics/<project>/sampleTopic"));
PCollection<KV<String, String>> keyedStream
浏览 1提问于2017-08-04得票数 4
回答已采纳
1回答
当前的波束管道正在使用FileIO.matchAll().continuously()以流的形式读取文件。这将返回PCollection。我想将这些具有相同名称的文件写回另一个gcs bucket，也就是说，每个PCollection都是一个文件metadata/readableFile，它应该在经过一些处理后写回另一个bucket。有没有什么接收器可以用来将每个PCollection项目写回GCS，或者有什么方法可以做到？是否有可能为每个元素创建一个窗口，然后使用一些GCS接收器IO来实现这一点。当操作一个窗口(即使它有多个元素)时，beam是否保证一个窗口要么被完全处理，要么根本不被处理，
浏览 6提问于2018-01-13得票数 1
我正在尝试找出服务帐户将文件写入GCS存储桶所需的最小权限集。我想我可以使用storage.objectAdmin，但是使用为Python SDK提供的示例代码失败，并显示一个错误：my-service-account@my-project.googleapis.com does not have storage.buckets.get access to my-bucket
client = _get_storage_client()
bucket = client.bucket(current_app.config['CLOUD_STORAGE_BUCKET'])
blob
浏览 7提问于2018-12-17得票数 4
尝试使用cloud composer访问存储桶时出现错误... 2019-03-18 11:50:00,651 {models.py:1594}错误- 404 GET https://www.googleapis.com/storage/v1/b/gs://xxxx-cloud-composer?projection=noAcl：未找到 def Ian_Log_Message():
from google.cloud import storage
import logging
logging.info('Hello Ian
浏览 11提问于2019-03-18得票数 0
我们的管道是这样的：
GCS (gz压缩文件) -> ParDo -> BigQuery
我想使用一个“扁平化”从GCS获取多个文件作为我的管道的输入。但它一直在犯错误：
Workflow failed. Causes: (5001e5764f46ac2c): BigQuery creation of import job for table "Impressions_05_2015_denormalized_test" in dataset "CPT_XXXX" in project "gdfp-XXXX" failed. Caus
浏览 5提问于2015-05-13得票数 4
回答已采纳
我正在尝试使用将谷歌PubSub消息写入。我知道TextIO/AvroIO不支持流式管道。然而，我在第一篇中读到，可以在流管道中从ParDo/DoFn写入GCS，这是作者在评论中提到的。我尽可能地紧跟他们的文章，建造了一条管道。
我的目的是为了这种行为：
以多达100的批次将消息写入GCS中的对象(每个窗口窗格一条)，路径对应于在dataflow-requests/[isodate-time]/[paneIndex]中发布消息的时间。
我得到了不同的结果：
每小时窗口中只有一个窗格。因此，我每小时只获得一个文件“桶”(它实际上是GCS中的一个对象路径)。将MAX_EVENTS_
浏览 1提问于2016-04-08得票数 17
回答已采纳
我试图从云存储桶中读取csv，并将其存储在pcollection中。为了使用桶进行身份验证，我使用了一个带有roles/storage.admin的服务帐户和一个JSON密钥。这是我的pipelinesOptions对象。
DataflowPipelineOptions dfOptions = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
dfOptions.setProject("project_name");
dfOptions.setStagingLocation(
浏览 16提问于2022-09-15得票数 0
回答已采纳
我正在尝试使用服务帐户密钥在Google Cloud Composer中设置Google Cloud Platform连接。因此，我创建了一个GCS存储桶，并将服务帐户密钥文件放入该存储桶中。密钥存储在JSON中。在keyfile path字段中，我指定了一个GCS bucket，在keyfile JSON字段中，我指定了文件名。作用域是。
在尝试使用此连接启动Dataproc集群时，我收到错误消息:找不到JSON文件。
查看错误消息，代码尝试使用：with open(filename, 'r') as file_obj解析文件，这显然不适用于GCS bucket路径。
所以我
浏览 2提问于2018-06-06得票数 6
2回答
我在datalab“玩”apache /数据流。我正在尝试从gcs读取csv文件。当我使用以下方法创建pcollection时：
lines = p | 'ReadMyFile' >> beam.io.ReadFromText('gs://' + BUCKET_NAME + '/' + input_file, coder='StrUtf8Coder')
我得到以下错误：
LookupError: unknown encoding: "THE","NAME","OF",&#
浏览 0提问于2018-08-20得票数 2
回答已采纳
1回答
桶的坏凭据
、、
我正在让Bad credentials for bucket尝试使用gcs引信安装一个GCS桶。我所做的：
创建了一个桶“mac-gcs-引信”
在同一个项目中创建具有所有存储权限的服务帐户。
运行以下命令
gcsfuse --key-file=/Users/tiagogouvea/Keys/gcp-Storage-Pessoal-mac-gcs-fuse-service-account.json mac-gcs-fuse ~/mac-gcs-fuse
我得到了：
Using mount point: /Users/tiagogouvea/mac-gcs-fuse
Opening GCS
浏览 0提问于2019-08-10得票数 2
我尝试使用新的谷歌BigQuery存储写入API在一个数据流作业中使用Beam。我在用
BigQueryIO<Pair<String,String>>write().withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
然而，当我运行它时，我会看到一个错误：
When writing an unbounded PCollection via FILE_LOADS or STORAGE_API_WRITES, triggering frequency must be specified
然而，用于的beam文档( tr
浏览 10提问于2022-02-10得票数 0
1回答
我有一个火花流应用程序，它从公共主题(例如kafka)读取消息，对每个消息应用一些转换，并将它们保存为GCS中由任意列划分的拼花文件。使用结构化流和火花-gcs连接器是相对容易的。例如，每条消息如下所示：
{
"app_id": "app1",
"user_id": "u001",
"evt_timestamp": 1617105047,
"evt_data": { ... }
}
我将其读取为结构化流DataFrame，然后通过例如app_id和user_id对其进行分区，然
浏览 5提问于2021-03-30得票数 2
回答已采纳
我正试图使用服务帐户运行一个Dataflow作业。我使用的是GcpOptions标志：
--serviceAccountKeyfile="dataflow-service-account.p12" --serviceAccountName="dataflow"
我收到以下错误：
Unable to verify that GCS bucket exists.com.google.cloud.dataflow.sdk.util.DataflowPathValidator.verifyPathIsAccessible(DataflowPathValidator.ja
浏览 0提问于2017-03-01得票数 3
我有一个云函数(CF1)，它反过来调用基于http的云函数(CF2)。
对于调用CF2，我使用"AuthorizedSession“，如下所示。在凭据变量中，我们是否可以将服务帐户作为身份验证的一部分来传递？
我不想存储服务帐户相关的信息，如服务帐户json文件，任何地方的gcp-项目。下面是一个例子：
credentials = service_account.IDTokenCredentials.from_service_account_file(service_account_info_file,
浏览 17提问于2022-07-12得票数 0
我希望能从Colab应用程序获得访问我的文件的正确权限设置的帮助。
目标
我希望能够从Colab python笔记本上访问CGS桶中的个人图像，该笔记本运行Tensorflow的“任意样式的样式传输”演示。
Situation
我设置了一个GCS桶，将其公之于众，并能够检索文件并在演示中使用它们。
为了避免GCS桶被公开访问，我删除了allUsers，并更改了与Colab和GCS绑定的帐户/电子邮件。
这会导致以下错误消息：
错误消息
Exception: URL fetch failure on https://storage.googleapis.com/01_bucket-02/Port
浏览 19提问于2022-03-15得票数 1
回答已采纳
1回答
使用 from spotify为Dataflow编写作业，下面的两个示例和将PubSub流写入GCS，但下面的代码会出现以下错误
误差
Exception in thread "main" java.lang.IllegalArgumentException: Write can only be applied to a Bounded PCollection
码
object StreamingPubSub {
def main(cmdlineArgs: Array[String]): Unit = {
// set up example wiring
val (opt
浏览 9提问于2016-10-05得票数 4
回答已采纳
我创建了一个Google服务帐户$GCP_SERVICE_ACCOUNT，它具有存储管理( Storage，roles/storage.admin)角色。
现在，我希望限制这个帐户，以便它只能访问特定的Google (GCS)桶($GCS_BUCKET_NAME)。
现在的问题是，$GCP_SERVICE_ACCOUNT可以访问所有的 GCS存储桶。我无法从其他GCS存储桶中删除$GCP_SERVICE_ACCOUNT，因为roles/storage.admin是继承的。
我该怎么办？
浏览 5提问于2019-12-04得票数 14
2回答
我的beam数据流作业在本地成功(使用DirectRunner)，在云上失败(使用DataflowRunner) 此代码片段中的本地化问题： class SomeDoFn(beam.DoFn):
...
def process(self, gcs_blob_path):
gcs_client = storage.Client()
bucket = gcs_client.get_bucket(BUCKET_NAME)
blob = Blob(gcs_blob_path, bucket)
# NEXT LINE IS CAUSING ISSUES! (w
浏览 19提问于2020-01-02得票数 1
回答已采纳
我有一个KV的值，其中PCollection是gcs file_patterns，值是文件的一些附加信息(例如，生成文件的“源”系统)。例如， KV("gs://bucket1/dir1/*", "SourceX"),
KV("gs://bucket1/dir2/*", "SourceY") 我需要一个PTransferm来将file_patterns扩展到GCS文件夹中的所有匹配文件，并保留"Source“字段。例如，如果dir1下有两个文件X1.dat，X2.dat，dir2下有一个文件(Y1.dat)，则输出为：
浏览 9提问于2019-02-23得票数 2
2回答
我需要通过java从本地或gcs读取avro文件。我遵循了文档中的示例
Pipeline p = ...;
// A Read from a GCS file (runs locally and using remote execution):
Schema schema = new Schema.Parser().parse(new File("schema.avsc"));
PCollection<GenericRecord> records =
p.apply(AvroIO.readGenericRecords(schema)
浏览 5提问于2018-03-03得票数 0
gsutil -m rm gs://{our_bucket}/{dir}/{subdir}/*
...
Removing gs://our_bucket/dir/subdir/staging-000000000102.json...
Removing gs://our_bucket/dir/subdir/staging-000000000101.json...
CommandException: 103 files/objects could not be removed.
该命令能够找到包含103个.JSON文件的目录，并在输出的Removing gs://...中“尝试”删除它们。我们收
浏览 4提问于2021-08-03得票数 1
我正在尝试编写一个数据存储流，用于读取pub sub的流，并写入大型查询。
在尝试运行该工具时，我收到错误“Unable to create a side-input view from input”和堆栈跟踪：
Exception in thread "main" java.lang.IllegalStateException: Unable to create a side-input view from input
at com.google.cloud.dataflow.sdk.transforms.View$AsIterable.validate(View.java:
浏览 1提问于2016-08-03得票数 1
回答已采纳
我用以下Node.js代码创建了一个文件：
const {Storage} = require('@google-cloud/storage')
var gcs = new Storage()
var bucket = gcs.bucket('bucket-name')
const file = bucket.file('filename')
// fileData is a utf8 buffer
file.save(fileData, function(err) {
console.log('Error:' + e
浏览 1提问于2019-05-22得票数 0
1回答
我正在尝试构建一个数据流模板。
目标是读取ValueProvider，它将告诉我要读取哪些文件。然后，对于每个文件，我需要使用对象来读取和丰富数据。我试过这样做：
p.apply(Create.of(options.getScheduleBatch()))
.apply(ParDo.of(StringScheduleBatchToFileReceivedFn.of()))
.apply(ParDo.of(new DoFn<FileReceived, PCollection<EventRow>>() {
浏览 1提问于2018-11-05得票数 0
回答已采纳
2回答
我试图使用bq命令行工具从GCS将数据加载到BigQuery中，并收到以下错误消息：
BigQuery error in load operation: Access Denied: Job mythical-maxim-293:bqjob_r11765e0cd9ceb52b_000001427694f0e1_1: RUN_JOB
我使用服务帐户(带有私钥)进行身份验证。我遵循了以下授予服务帐户访问级别的链接：
服务帐户电子邮件被授予了对BigQuery数据集的写访问权和使用GCS桶的读访问权。
注:添加服务帐户电子邮件作为一个项目的作者，解决了问题，但这是不可行的情况下。我不允许请求项目
浏览 4提问于2013-11-20得票数 1
回答已采纳
我试图使用我的服务帐户在本地使用的简单代码将图像上传到：
const storage = require('@google-cloud/storage');
const fs = require('fs');
const gcs = storage({
projectId: 'ID',
keyFilename: 'KEYNAME'
});
var bucket = gcs.bucket('BUCKET NAME');
bucket.upload('hiking-image.jpg'
浏览 4提问于2016-11-25得票数 1
回答已采纳
1回答
当用户删除他们的帐户时，我希望删除他们的存储文件和数据。
我能够为RTDB执行多路径删除，我如何做到这一点，同时也从存储中删除文件？
我试过在.then上链接，但它使所有东西都失败了……
例如..。
.then(() => {
const bucket = gcs.bucket(functions.config().firebase.storageBucket);
const path = `categories/${uid}`;
return bucket.file(path).delete();
})
我希望在不总是部署的情况下测试你的功能会更快，因为它已经花费了太多的时
浏览 0提问于2019-08-16得票数 0
我一直在努力处理一个需要发出外部输出的作业，因为我一直得到一个异常(“无法序列化xxx”)。
尽管我显式地为我所使用的类型指定了一个编码器，但我仍然得到了相同的错误，所以我决定按照以下文档编写一个简单的工作：
令我惊讶的是，我仍然得到了同样的例外，现在我怀疑我一定做错了什么(但我自己也搞不清楚)。就代码而言，我试图遵循上面给出的示例。
下面，我发布了源代码以及运行时收到的错误消息。我相信这是可复制的(将“GCS_BUCKET”更改为您拥有的任何桶，并创建main()方法，使用args调用'TestSideOutput‘)，最好知道其他人是否可以在其端进行复制。我们使用的是JDK 8和
浏览 1提问于2016-10-04得票数 1
回答已采纳
我在google云上创建一个接收器，目标作为发布/子主题，用于它的命令如下所示。gcloud日志接收器创建geo_sinktest
网站/projects//projects/
-log-filter‘( (protoPayload.methodName="storage.objects.create“”gcs_bucket“和resource.type= OR protoPayload.methodName="storage.objects.delete"))或resource.type=”bigquery_dataset“)
在执行命令后，当我单击“查看接收器”时
浏览 0提问于2021-05-11得票数 0
回答已采纳
2回答
我正在使用Django -gcloud存储库来允许我的Django应用程序在GCS上存储图像。
我已经创建了一个桶，GCS服务帐户，并根据下载了一个JSON密钥文件。但是，将映像保存到云存储会引发以下错误：
AssertionError at /showroom/create/
Credentials file not found
...
/srv/showroom/views.py in form_valid
images.save() …
▶ Local vars
...
/env/lib/python3.7/site-packages/django_g
浏览 4提问于2019-10-27得票数 1
最近，我遇到了一个问题，当我使用pyspark将数据写入BigQuery中时。这里是：
pyspark.sql.utils.IllegalArgumentException:必须通知U‘informed或持久性GCS桶
在对问题进行研究后，我发现暂时性的GCS桶要提到spark.conf。
bucket = "temp_bucket"
spark.conf.set('temporaryGcsBucket', bucket)
我认为在Biquery中为像Hive这样的表创建一个文件是没有概念的。
我想知道更多关于它的情况，为什么我们需要临时gcs将数据写入
浏览 6提问于2021-06-11得票数 1
回答已采纳
2回答
从昨天起我就一直在为这个问题而奋斗。寻找帮助其他面临类似问题的人的解决方案，但对我没有任何帮助。如果有人能帮我，我会很感激的！
我已经创建了一个项目(名称: xxx)，它使用app引擎(id: app-xxx)监视一个GCS桶(名称: bucket-xxx)。App引擎和桶在同一个项目中。我试着按照列出的步骤：
使用p12键创建服务帐户(电子邮件: xxx@app-xxx.iam.gserviceaccount.com)。
白色URL (属性: https: //app-xxx.apppot.com)。我看到网站管理员搜索控制台中列出的URL。当我使用webmaster list AP
浏览 4提问于2015-12-11得票数 2
回答已采纳
这个问题延续了以前的一篇文章的讨论：
总之，我希望转换一些文件，但希望它们的输入和输出匹配。例如：
gcs_files: ['input/gcs_file_1.json', 'input/gcs_file_2.json', 'input/gcs_file_3.json']
Output: ['output/gcs_file_1_transformed.json', 'output/gcs_file_2_transformed.json', 'output/gcs_file_2_transformed.j
浏览 12提问于2022-08-25得票数 0
1回答
我遵循这个在Google上进行培训。我一步一步地遵循它，但是当我把ml作业提交到云端时，我面临着错误。我执行了这个命令。
sam@ submit：~/models/research$ gcloud ml-引擎作业提交培训whoami_object_detection_date +%s --作业-dir=gs://tf_dist/train packages dist/object_detection-0.1.tar.gz，slim/dist/slim 0.1.tar.gz --模块名称object_detection.train -us区域-集中区-配置对象_检测/样本/云/Cloud.ym
浏览 1提问于2017-12-07得票数 3
1回答
我创建了桶，上传了一个文件
gsutil cp -a public-read test.htm gs://BUCKETNAME
我可以在存储浏览器中看到该文件，并使用gsutil ls
当我尝试用下面的代码阅读它时，我会得到
NotFoundError:期望状态200来自Google。但得到404状态。
代码：
import logging
import os
import lib.cloudstorage as gcs
import webapp2
from google.appengine.api import app_identity
my_default_retry_p
浏览 1提问于2017-09-24得票数 0
回答已采纳
我已经创建了一个具有roles/storage.admin角色的服务帐户，其条件如下：
resource.name.startsWith(
"//storage.googleapis.com/projects/_/buckets/rakibs-new-bucket"
)
但是，当此服务帐户试图查询桶中的任何内容时，它会如下所示：
Error: Failed to get existing workspaces: querying Cloud Storage failed: googleapi: Error 403: my-new-service-account@my-pro
浏览 2提问于2020-04-03得票数 0
回答已采纳
我正在执行以下步骤：https://docs.snowflake.com/en/user-guide/data-load-snowpipe-auto-gcs.html，并且在步骤6中遇到了问题 我跑了一下： create storage integration my_integration
type = external_stage
storage_provider = gcs
浏览 23提问于2021-08-20得票数 0
1回答
使用元素计数用数据流写入GCS
、、、、
这是针对Apache 2.2.0版本的。
我正在尝试使用AfterPane.elementCountAtLeast(...)，但到目前为止还没有取得任何成功。我想要的东西看起来很像，但需要适应2.2.0。最终，我只需要一个简单的或，在X元素或Y时间过去后写一个文件。我打算将时间设置得非常高，以便在大多数情况下对元素的数量进行写入，并且只在消息量非常低的时候根据持续时间进行写入。
使用作为参考，以下是我尝试过的：
String bucketPath =
String.format("gs://%s/%s",
options.getBucketName()
浏览 2提问于2018-01-12得票数 0
回答已采纳
2回答
在应用程序引擎中，我可以使用应用程序的默认桶提供云存储文件，比如pdf：
http://storage.googleapis.com/<appid>.appspot.com/<file_name>
但是，如何在不使用blob_key的情况下为SDK中的本地云存储文件提供服务呢？
我这样写到默认的桶中：
gcs_file_name = '/%s/%s' % (app_identity.get_default_gcs_bucket_name(), file_name)
with gcs.open(gcs_file_name, 'w') as
浏览 2提问于2014-03-04得票数 9
回答已采纳
我目前正在创建一个PCollectionView，方法是从gcs桶中读取过滤信息，并将它作为侧输入传递到管道的不同阶段，以便过滤输出。如果gcs桶中的文件发生了更改，我希望当前正在运行的管道使用这个新的过滤器信息。如果筛选器发生变化，是否有方法在每个新的数据窗口上更新此PCollectionView？我以为我可以在startBundle上做这件事，但我不知道它是如何做到的，也不知道它是否可行。如果可能的话，你能举个例子吗？
PCollectionView<Map<String, TagObject>>
tagMapView =
pipeline
浏览 2提问于2016-12-21得票数 2
回答已采纳
我正在尝试让用户直接从我在AppEngine Flex (PHP7.2)上的应用程序上传到谷歌云存储。代码在本地工作(不使用GCP本地开发服务器)，但是当我将代码上传到AppEngine并打开页面/视图时，我得到了一个500错误。想知道有没有更有GCP和GCS经验的人能指出我做错了什么？ 我将调试语句(为简洁起见将其删除)放入实时服务器代码中，在调用$object->beginSignedUploadSession() (请参见下面的代码)之前，我可以看到这些语句立即停止。 $config = [
'projectId' => 'my-projec
浏览 11提问于2019-06-14得票数 0
2回答
新鲜的气流。我试图将结果保存到另一个桶(而不是气流桶)中的文件中。我可以保存到'/home/airflow/gcs/data/test.json‘中的一个文件，然后使用gcs_hook.GoogleCloudStorageHook复制到另一个桶中。以下是代码：
def write_file_func(**context):
file = f'/home/airflow/gcs/data/test.json'
with open(file, 'w') as f:
f.write(json.dumps('{"n
浏览 2提问于2019-12-16得票数 2
回答已采纳
1回答
在学习教程的同时，我已经到了在云上训练模型的阶段。不幸的是，命令
gcloud ml-engine jobs submit training `whoami`_object_detection_`date +%s` \
--job-dir=gs://${YOUR_GCS_BUCKET}/train \
--packages dist/object_detection-0.1.tar.gz,slim/dist/slim-0.1.tar.gz \
--module-name object_detection.train \
--region us-central1
浏览 1提问于2017-10-03得票数 1
回答已采纳
1回答
我有一个场景，需要执行以下操作：
从GCS中存储的文件中读取数据
将多个转换应用于数据。
在Google中持久化PCollection。
当步骤3成功时，将PCollection写到一个卡夫卡主题。
我的问题是如何在Apache中实现它。我的主要要求是，只有在“插入大查询”成功的情况下，我才能写信给Kafka。
我发现了类似的东西：
其中的数据被写入到全球控制系统和烧烤。但我的要求是，只有在写到BQ成功之后，才能给卡夫卡写信。有没有人知道这是可能的，我是如何做到的？
浏览 4提问于2022-07-06得票数 0
2回答
我知道如何从云运行实例中的云存储下载文件。但是，我找不到在python中读取文件的语法。我希望立即将csv文件转换为熊猫数据，只需使用pd.read_csv('testing.csv')即可。所以我的个人代码看起来像，download_blob(bucket_name, source_blob_name, 'testing.csv')。那么，我不应该能够在云运行实例中执行pd.read_csv('testing.csv')吗？这样做的时候，我总是在加载页面时得到一个内部服务器。这似乎是一个简单的问题，但我在任何地方都找不到它的例子。所有的东西都只
浏览 13提问于2021-11-26得票数 2
回答已采纳
1回答
我是云领域的初学者，希望将我的dataproc cluster访问限制在项目中给定的gcs buckets上。
让我们说我已经创建了一个名为service account的'data-proc-service-account@my-cloud-project.iam.gserviceaccount.com'，然后我创建了一个dataproc集群并将服务帐户分配给它。
现在，我已经创建了两个名为
'gs://my-test-bucket/spark-input-files/'
'gs://my-test-bucket/spark-output-files/
浏览 6提问于2020-07-29得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
