Похоже, проблема в том, что мы возвращаем разное количество строк по умолчанию для query() и getQueryResults(). Таким образом, в зависимости от того, быстро ли завершился ваш запрос (и поэтому вам не нужно было использовать getQueryResults()), вы либо получили больше, либо меньше строк.
Я зарегистрировал ошибку, и скоро мы должны ее исправить.
Обходной путь (и в целом хорошая идея) состоит в том, чтобы установить maxResults как для запроса, так и для вызовов getQueryResults. И если вам нужно много строк, вы можете просмотреть результаты, используя возвращаемый токен страницы.
Ниже приведен пример чтения одной страницы данных из завершенного задания запроса. Он будет включен в следующий выпуск bq.py:
class _JobTableReader(_TableReader):
"""A TableReader that reads from a completed job."""
def __init__(self, local_apiclient, project_id, job_id):
self.job_id = job_id
self.project_id = project_id
self._apiclient = local_apiclient
def ReadSchemaAndRows(self, max_rows=None):
"""Read at most max_rows rows from a table and the schema.
Args:
max_rows: maximum number of rows to return.
Raises:
BigqueryInterfaceError: when bigquery returns something unexpected.
Returns:
A tuple where the first item is the list of fields and the
second item a list of rows.
"""
page_token = None
rows = []
schema = {}
max_rows = max_rows if max_rows is not None else sys.maxint
while len(rows) < max_rows:
(more_rows, page_token, total_rows, current_schema) = self._ReadOnePage(
max_rows=max_rows - len(rows),
page_token=page_token)
if not schema and current_schema:
schema = current_schema.get('fields', {})
max_rows = min(max_rows, total_rows)
for row in more_rows:
rows.append([entry.get('v', '') for entry in row.get('f', [])])
if not page_token and len(rows) != max_rows:
raise BigqueryInterfaceError(
'PageToken missing for %r' % (self,))
if not more_rows and len(rows) != max_rows:
raise BigqueryInterfaceError(
'Not enough rows returned by server for %r' % (self,))
return (schema, rows)
def _ReadOnePage(self, max_rows, page_token=None):
data = self._apiclient.jobs().getQueryResults(
maxResults=max_rows,
pageToken=page_token,
# Sets the timeout to 0 because we assume the table is already ready.
timeoutMs=0,
projectId=self.project_id,
jobId=self.job_id).execute()
if not data['jobComplete']:
raise BigqueryError('Job %s is not done' % (self,))
page_token = data.get('pageToken', None)
total_rows = int(data['totalRows'])
schema = data.get('schema', None)
rows = data.get('rows', [])
return (rows, page_token, total_rows, schema)
person
Jordan Tigani
schedule
19.07.2013