Tiene un tipo de almacén de datos GAE con varios 100'000 de objetos en ellos. Desea realizar varias consultas relacionadas (que implican consultas de conteo). Big Query parece un dios apto para hacer esto.Google App Engine: ¿usa Big Query en el almacén de datos?
¿Existe actualmente una manera fácil de consultar un almacén de datos de AppEngine en vivo usando Big Query?
No se puede ejecutar un BigQuery directamente en las entidades del DataStore, pero puede escribir un Pipeper Pipeline que lea entidades de DataStore, las escriba en CSV en Google Cloud Storage y luego las incorpore a BigQuery; incluso puede automatizar el proceso. He aquí un ejemplo del uso de los Mapper API clases por sólo el almacén de datos a CSV paso:
import re
import time
from datetime import datetime
import urllib
import httplib2
import pickle
from google.appengine.ext import blobstore
from google.appengine.ext import db
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.ext.webapp import blobstore_handlers
from google.appengine.ext.webapp import util
from google.appengine.ext.webapp import template
from mapreduce.lib import files
from google.appengine.api import taskqueue
from google.appengine.api import users
from mapreduce import base_handler
from mapreduce import mapreduce_pipeline
from mapreduce import operation as op
from apiclient.discovery import build
from google.appengine.api import memcache
from oauth2client.appengine import AppAssertionCredentials
#Number of shards to use in the Mapper pipeline
SHARDS = 20
# Name of the project's Google Cloud Storage Bucket
GS_BUCKET = 'your bucket'
# DataStore Model
class YourEntity(db.Expando):
field1 = db.StringProperty() # etc, etc
ENTITY_KIND = 'main.YourEntity'
class MapReduceStart(webapp.RequestHandler):
"""Handler that provides link for user to start MapReduce pipeline.
"""
def get(self):
pipeline = IteratorPipeline(ENTITY_KIND)
pipeline.start()
path = pipeline.base_path + "/status?root=" + pipeline.pipeline_id
logging.info('Redirecting to: %s' % path)
self.redirect(path)
class IteratorPipeline(base_handler.PipelineBase):
""" A pipeline that iterates through datastore
"""
def run(self, entity_type):
output = yield mapreduce_pipeline.MapperPipeline(
"DataStore_to_Google_Storage_Pipeline",
"main.datastore_map",
"mapreduce.input_readers.DatastoreInputReader",
output_writer_spec="mapreduce.output_writers.FileOutputWriter",
params={
"input_reader":{
"entity_kind": entity_type,
},
"output_writer":{
"filesystem": "gs",
"gs_bucket_name": GS_BUCKET,
"output_sharding":"none",
}
},
shards=SHARDS)
def datastore_map(entity_type):
props = GetPropsFor(entity_type)
data = db.to_dict(entity_type)
result = ','.join(['"%s"' % str(data.get(k)) for k in props])
yield('%s\n' % result)
def GetPropsFor(entity_or_kind):
if (isinstance(entity_or_kind, basestring)):
kind = entity_or_kind
else:
kind = entity_or_kind.kind()
cls = globals().get(kind)
return cls.properties()
application = webapp.WSGIApplication(
[('/start', MapReduceStart)],
debug=True)
def main():
run_wsgi_app(application)
if __name__ == "__main__":
main()
Si añade esto al final de su clase IteratorPipeline: yield CloudStorageToBigQuery(output), se puede canalizar el gestor de archivo csv resultante en un tubo de la ingestión de BigQuery. .. así:
class CloudStorageToBigQuery(base_handler.PipelineBase):
"""A Pipeline that kicks off a BigQuery ingestion job.
"""
def run(self, output):
# BigQuery API Settings
SCOPE = 'https://www.googleapis.com/auth/bigquery'
PROJECT_ID = 'Some_ProjectXXXX'
DATASET_ID = 'Some_DATASET'
# Create a new API service for interacting with BigQuery
credentials = AppAssertionCredentials(scope=SCOPE)
http = credentials.authorize(httplib2.Http())
bigquery_service = build("bigquery", "v2", http=http)
jobs = bigquery_service.jobs()
table_name = 'datastore_dump_%s' % datetime.utcnow().strftime(
'%m%d%Y_%H%M%S')
files = [str(f.replace('/gs/', 'gs://')) for f in output]
result = jobs.insert(projectId=PROJECT_ID,
body=build_job_data(table_name,files)).execute()
logging.info(result)
def build_job_data(table_name, files):
return {"projectId": PROJECT_ID,
"configuration":{
"load": {
"sourceUris": files,
"schema":{
# put your schema here
"fields": fields
},
"destinationTable":{
"projectId": PROJECT_ID,
"datasetId": DATASET_ID,
"tableId": table_name,
},
}
}
}
No, BigQuery es un producto diferente que necesita los datos que se cargarán en él. No puede funcionar sobre el almacén de datos. Puede usar GQL para consultar el almacén de datos.
Para BigQuery tienes que exportar esos tipos a una estructura de registro CSV o delimitada, cargar en BigQuery y puedes consultar. No hay ninguna instalación que conozca que permita consultar el Datastore de GAE en vivo.
Biquery es un motor de consultas analíticas que significa que no puede cambiar el registro. No se permiten actualizaciones ni eliminaciones, solo se puede agregar.
Estamos haciendo un programa de testers de confianza para pasar del almacén de datos a BigQuery en dos operaciones simples:
Se encarga automáticamente del esquema para usted.
Más información (aplicar): https://docs.google.com/a/google.com/spreadsheet/viewform?formkey=dHdpeXlmRlZCNWlYSE9BcE5jc2NYOUE6MQ
Con el nuevo (a partir de septiembre de 2013) streaming inserts api puede importar los registros de su aplicación en BigQuery.
Los datos están disponibles en BigQuery inmediatamente, por lo que esto debería satisfacer sus requisitos en tiempo real.
Mientras que esta pregunta es ahora un poco viejo, esto puede ser una solución más fácil para cualquier persona tropezarse con esta pregunta
Por el momento, aunque cada vez que esto funcione desde un servidor dev local es irregular en el mejor.
A partir de 2016, esto es muy posible ahora! Usted debe hacer lo siguiente:
¡Vea this post para obtener un ejemplo completo de este flujo de trabajo!
Entonces, ¿qué pasó con esto? ¿Alguna actualización sobre el destino de TTP? – gae123
sí, ha sido un tiempo – ZiglioUK
También le interesó – Omri