使用消息数据上传到大查询-谷歌数据流
我正在通过pub-sub接收消息，并希望使用消息数据上传到big-query以确定要上传数据到什么表。
我尝试执行以下操作：
Pipeline.create（选项）；字符串BigQueryTable；
PCollection<String> input = pipeline
.apply(PubsubIO.Read.subscription("projects/my-data-analysis/subscriptions/myDataflowSub"));
input.apply(ParDo.of(new DoFn<String, TableRow>() {
@Override
public void processElement(DoFn<String, TableRow>.ProcessContext c) throws Exception {
JSONObject firstJSONObject = new JSONObject(c.element());
bigQueryTable = firstJSONObject.get("tableName").toString();
TableRow tableRow = convertJsonToTableRow(firstJSONObject);
c.output(tableRow);
}
})).apply(BigQueryIO.Write.to("my-data-analysis:mydataset." + bigQueryTable).withSchema(tableSchema));
有没有什么方法可以做到这一点，而不编写我自己的DOFN？
如果我确实需要实现我自己的doFn，如何实现它上传到大查询？
共有1个答案
-
谷歌地图 LSV默认的对谷歌影像进行了加载，如果需要加载其他的谷歌地图数据，可以通过LSV中直接点击即可加载。 谷歌历史影像 LSV中支持对谷歌历史影响的查看，。通过历史影像，我们可以分析出很多单一影像无法获取的信息。触摸历史，寻找其他时空中发生的事情和变迁。历史影像主要用于进行多时相对比、去云、多角度观察、小路寻找等。 开启前： 开启后：
-
谷歌地图 LSV默认的对谷歌影像进行了加载，如果需要加载其他的谷歌地图数据，可以通过LSV中直接点击即可加载。 谷歌历史影像 LSV中支持对谷歌历史影响的查看，。通过历史影像，我们可以分析出很多单一影像无法获取的信息。触摸历史，寻找其他时空中发生的事情和变迁。历史影像主要用于进行多时相对比、去云、多角度观察、小路寻找等。 开启前： 开启后：
-
我正在尝试使用Google Cloud Dataflow（Python SDK）将Google PubSub消息写入Google Cloud Storage。这些消息以json格式进入PubSub，我必须定义一个模式才能将它们写入Google Cloud Storage中的拼花格式。 根据其他用户的建议，我开始着手这项任务，特别是查看这个和这个来源。< br >第一个不是我想要做的，因为它将更改应
-
我正试图使用Python应用编程接口将文件从笔记本中的数据实验室实例上传到我的谷歌存储桶，但我无法解决这个问题。谷歌在其留档中提供的代码示例似乎在数据实验室中不起作用。我目前正在使用gsutil命令，但想了解如何在使用Python应用编程接口时做到这一点。 文件目录（我想上传checkpoints文件夹中的python文件）： 当前代码： 目前的工作解决方案：
-
我正在尝试使用谷歌云数据流将谷歌PubSub消息写入谷歌云存储。PubSub消息采用json格式，我要执行的唯一操作是从json到parquet文件的转换。
-
当我对一大群小对象（15k只有几个短字符串和布尔属性的对象）运行查询时，没有对这些对象做任何事情，我看到我的实例的内存使用量不断增加（增加了70Mb）。内存的增加看起来与查询所需的内存数据量不成比例。 我使用的循环如下所示： 为了确保这不是由于appstats造成的，我调用不记录任何统计数据。 有人知道会发生什么吗？或者关于如何调试/配置文件的任何提示？ 更新1：我打开了，我看到垃圾收集器被定期调