小编Gra*_*ley的帖子

我们有一个从 BigQuery 表读取的管道。调用getF()TableRow 上的方法应该List<TableCell>为该行返回 a 。但它返回null。

为什么getF()返回null？

@Override
public void processElement(ProcessContext c) throws Exception {
   TableRow aRow = context.element();
   List<TableCell> tableCells = aRow.getF(); //This returns null! 
}

我有一张名为：结果的表 我正在使用 BigQuery 从 GA 中选择数据

SELECT
  Date,
  totals.pageviews,
  h.transaction.transactionId,
  h.item.itemQuantity,
  h.transaction.transactionRevenue,
  totals.bounces,
  fullvisitorid,
  totals.timeOnSite,
  device.browser,
  device.deviceCategory,
  trafficSource.source,
  channelGrouping,
  h.page.pagePath,
  h.eventInfo.eventCategory,
  device.operatingSystem
FROM
  `atomic-life-148403.126959513.ga_sessions_*`,
  UNNEST(hits) AS h
WHERE
  _TABLE_SUFFIX BETWEEN REPLACE(CAST(DATE_ADD(CURRENT_DATE(), INTERVAL -1 YEAR) AS STRING), '-','')
  AND CONCAT('intraday_', REPLACE(CAST(DATE_ADD(CURRENT_DATE(), INTERVAL 0 DAY) AS STRING), '-',''))
  ORDER BY
  date  DESC

有一些记录重复。如何删除表中的重复记录？

我想得到以下结果。

gsutil 有一个非常简洁的功能，称为“ rsync ”，可以同步 GCS 中的 2 个存储桶/目录。我搜索了JSON API和Java 客户端库，但看不到任何支持此类开箱即用功能的东西。

有谁知道如何通过Java客户端实现“rsync”的相同功能（无需实际编写两个存储桶之间的差异逻辑）？

使用SPLIT()&NTH(),我将分割一个字符串值,并将第二个子字符串作为结果.然后我想对结果进行分组.但是,当我将SPLIT()与GROUP BY结合使用时,它会不断给出错误:

Error: (L1:55): Cannot group by an aggregate

结果是一个字符串,为什么不能对它进行分组？

例如,这可以工作并返回正确的字符串:

SELECT NTH(2,SPLIT('FIRST-SECOND','-')) as second_part FROM [FOO.bar] limit 10

但是然后对结果进行分组不起作用:

SELECT NTH(2,SPLIT('FIRST-SECOND','-')) as second_part FROM [FOO.bar] GROUP BY second_part limit 10

我在 BigQuery 中有一个联合源，它指向 GCS 中的一些 CSV 文件。

当我尝试读取联合 BigQuery 表作为 Dataflow 管道的源时，它会引发以下错误：

    1226 [main] ERROR com.google.cloud.dataflow.sdk.util.BigQueryTableRowIterator  - Error reading from BigQuery table Federated_test_dataflow of dataset CPT_7414_PLAYGROUND : 400 Bad Request
{
  "code" : 400,
  "errors" : [ {
    "domain" : "global",
    "message" : "Cannot list a table of type EXTERNAL.",
    "reason" : "invalid"
  } ],
  "message" : "Cannot list a table of type EXTERNAL."
}

Dataflow 不支持 BigQuery 中的联合源，还是我做错了什么？我确实知道我可以将 GCS 中的文件直接读取到我的管道中，但TableRow由于应用程序的设计，我更喜欢使用 BigQuery对象。

 PCollection<TableRow> results = pipeline.apply("fed-test", BigQueryIO.Read.from("<project_id>:CPT_7414_PLAYGROUND.Federated_test_dataflow")).apply(ParDo.of(new DoFn<TableRow, …

我有一个Python脚本,它从firebase下载数据,操作它然后将其转储到JSON文件中.我可以通过命令行将其上传到BigQuery,但现在我想将一些代码放入Python脚本中以便将其全部完成.

这是我到目前为止的代码.

import json
from firebase import firebase

firebase = firebase.FirebaseApplication('<redacted>')
result = firebase.get('/connection_info', None)
id_keys = map(str, result.keys())

#with open('result.json', 'r') as w:
 # connection = json.load(w)

with open("w.json", "w") as outfile:
  for id in id_keys:
    json.dump(result[id], outfile, indent=None)
    outfile.write("\n")

我们一直在努力在（标准 sql）BigQuery 中循环数据，但没有成功。

我不确定它是否是 sql 支持的功能，我们对问题的理解，或者我们想要在 BigQuery 中执行此操作的方式。

无论如何，假设我们有一个事件表，其中每个事件都由用户 ID 和日期描述（同一用户 ID 在同一日期可能有许多事件）

id  STRING
dt  DATE

我们想知道的一件事是在给定的时间段内有多少不同的用户生成了事件。这是相当微不足道的，只是表上的一个 COUNT，以句点作为 WHERE 子句中的约束。例如，如果我们有四个月的时间段：

SELECT
  COUNT(DISTINCT id) AS total
FROM
  `events`
WHERE
  dt BETWEEN DATE_ADD(CURRENT_DATE(), INTERVAL -4 MONTH)
  AND CURRENT_DATE()

但是，如果我们希望在相同的给定时间段内递归地获取其他天（或周）的历史记录，就会出现问题。例如，昨天，前天，等等......直到......例如，3个月前。所以这里的变量将是 CURRENT_DATE() ，它可以回溯一天或任何一个因素，但间隔保持不变（在我们的例子中是 4 个月）。我们期待这样的事情（一天的因素）：

2017-07-14 2017-03-14 1760333
2017-07-13 2017-03-13 1856333
2017-07-12 2017-03-12 2031993
...
2017-04-14 2017-01-14 1999352

这只是对同一张桌子上的每一天、每周等进行循环，然后对这段时间内发生的不同事件进行计数。但是我们不能在 BigQuery 中进行“循环”。

One way we thought was a JOIN, and then a COUNT on the GROUP BY intervals (taking advantage of the HAVING clause …

我们编写了一个 Google 数据流代码，该代码将一个值插入到一个列是 DateTime 类型的 bigquery 表中。大多数时候逻辑运行良好。但是突然我们遇到了无效的日期时间问题。

Exception: java.lang.RuntimeException: java.io.IOException: Insert failed: [{"errors":[{"debugInfo":"generic::out_of_range: Invalid datetime string \"2017-09-26T21:16\"

目前尚不清楚上述值如何以及为何无效。我们看到它遵循https://cloud.google.com/bigquery/docs/reference/standard-sql/data-types 中提到的 DateTime 数据类型

此外，目前还不清楚为什么它只是偶尔抛出这个错误。

我们写了一个自定义的转换代码来扩展 DoFn ProcessElement 代码是这样的

 public void processElement(ProcessContext c) throws Exception {

    TableRow tableRow = c.element();
    try {
       // do some processing then 
      tableRow.set("PredictedDate",**LocalDateTime.now().toString()**);
      c.output(tableRow);
    }catch(Exception exc){
        LOG.error("Exception while processing and hence not attempting to write to bigquery");
    }
} 

enter code here

它工作正常，但在夜间（美国中部时区）偶尔会失败。你能帮我们找到根本原因吗？

我为这个天真的问题道歉,但我想澄清一下Cloud Dataflow或Cloud Composer是否适合这项工作,我不清楚Google文档.

目前,我正在使用Cloud Dataflow读取非标准csv文件 - 执行一些基本处理 - 并将其加载到BigQuery中.

让我举一个非常基本的例子:

# file.csv
type\x01date
house\x0112/27/1982
car\x0111/9/1889

从这个文件中我们检测到模式并创建一个BigQuery表,如下所示:

`table`
type (STRING)
date (DATE)

而且,我们还格式化我们的数据以插入(在python中)到BigQuery:

DATA = [
    ("house", "1982-12-27"),
    ("car", "1889-9-11")
]

这是对正在发生的事情的极大简化,但这就是我们目前使用Cloud Dataflow的方式.

那么我的问题是,Cloud Composer图片中的位置是什么？它可以在上面提供哪些附加功能？换句话说,为什么它会在"云数据流"之上使用？

我将 Cloud Build 与gcloud构建器一起使用。我将其覆盖，entrypoint以便bq可以在构建步骤中运行一些 BigQuery SQL。以前，我将 SQL 直接嵌入到 Cloud Build 的 YAML 配置中。这工作正常：

steps:
- name: gcr.io/cloud-builders/gcloud
  entrypoint: 'bq'
  args: ['query', '--use_legacy_sql=false', 'SELECT 1']

现在我想将 SQL 从 YAML 重构到一个文件中。根据这里，您可以将cat文件或管道传送到bq. 这可以在命令行上运行，没有任何问题。

但是，我无法让它与 Cloud Build 一起使用。我尝试了很多不同的组合，以及转义字符等，但无论我尝试什么，shell 都不会评估/执行反引号cat my_query.sl，而是认为它是查询本身：

工作正常：

在 Cloud Build 中构建它不会工作：

steps:
- name: gcr.io/cloud-builders/gcloud
  entrypoint: 'bq'
  args: ['query', '--use_legacy_sql=false', '`cat my_query.sql`']

我也尝试通过管道而不是使用cat，但我得到了同样的错误。

我一定在这里遗漏了一些明显的东西，但我看不到它。我可以构建一个自定义 docker 映像，并将所有内容包装在 shell 脚本中，但如果可能的话，我宁愿不必这样做。

如何在构建步骤中使用 Cloud Build 和 shell 评估？