HTTP-Batchanfragen für Data Catalog erstellen

Jede HTTP-Verbindung, die Ihre Anwendung erstellt, erfordert einen bestimmten Overhead. Data Catalog API-Anfragen unterstützen das Batching, sodass Sie mehrere API-Aufrufe in einer einzigen HTTP-Anfrage zusammenfassen können. Sie können die HTTP-Batchverarbeitung verwenden, falls Sie viele kleine Anfragen stellen und den Overhead der HTTP-Anfrage minimieren möchten. Beachten Sie: Obwohl das Batching den Overhead reduziert, werden Anfragen innerhalb eines Batches für API-Kontingente weiterhin als mehrere Anfragen gezählt.

Eine allgemeine Dokumentation zur Verwendung von HTTP-Batch mit Google Cloud findet sich in der Google API Python-Clientdokumentation.

HTTP-Batchanfragen in Python erstellen

Um Batchanfragen zum Erstellen oder Bearbeiten von Einträgen in Data Catalog zu nutzen, müssen Sie zuerst mit catalog.search() oder entries.lookup() nach den Einträgen suchen, die Sie ändern möchten.

Führen Sie folgende Schritte aus, um mit der Google Python API eine HTTP-Batchanfrage zu erstellen:

  1. Erstellen Sie ein BatchHttpRequest-Objekt, indem Sie new_batch_http_request() oder den BatchHttpRequest()-Konstruktor aufrufen. Sie können einen Callback übergeben, der als Reaktion pro Anfrage aufgerufen wird.
  2. Rufen Sie add() für das Objekt BatchHttpRequest für jede auszuführende Anfrage auf. Wenn Sie beim Erstellen Ihres BatchHttpRequest-Objekts einen Callback übergeben haben, kann jeder add() an den Callback zu übergebende Parameter enthalten.
  3. Nachdem Sie die Anfragen hinzugefügt haben, rufen Sie execute() für das Objekt BatchHttpRequest auf, um sie auszuführen. Die Funktion execute() wird blockiert, bis alle Callbacks aufgerufen wurden.

Anfragen in einem BatchHttpRequest können parallel ausgeführt werden. Es gibt keine Garantie für die Reihenfolge der Ausführung. Das bedeutet, dass Anfragen im selben Batch nicht voneinander abhängig sein dürfen. Beispiel: Sie sollten in derselben Anfrage nie gleichzeitig ein zusammengehöriges EntryGroup und ein Entry erstellen, da die Erstellung von Entry vor der Erstellung von EntryGroup erfolgen kann (was dazu führen würde, dass die Ausführung fehlschlägt).

Batchanfragen mit regionalen Endpunkten

Wenn Sie HTTP-Batchanfragen mit regionalen API-Endpunkten von Data Catalog verwenden, müssen alle API-Anfragen in einem Batch derselben Region angehören. Beim Ausführen des Batches müssen Sie den richtigen regionalen Endpunkt aufrufen. Wenn sich Ihre Ressourcen beispielsweise in us-central1 befinden, rufen Sie https://us-central1-datacatalog.googleapis.com/batch auf.

Regional unabhängige APIs

Regional unabhängige APIs (z. B. catalog.lookup() und entries.search()) können gruppiert werden, aber dürfen nicht mit regional abhängigen APIs gruppiert werden. Für regional unabhängige APIs verwenden Sie den Endpunkt https://datacatalog.googleapis.com/batch.

Beispiel

Diese Python-Beispielanwendung veranschaulicht die Verwendung einer HTTP-Batchanfrage, um mithilfe der Data Catalog API mehrere Tags aus einer Tag-Vorlage zu erstellen.

 
from googleapiclient.discovery import build
from googleapiclient.http import BatchHttpRequest
from oauth2client.service_account import ServiceAccountCredentials
import uuid

#-------------------------------------------------------------#
# 0. Helper and initialization logic
#-------------------------------------------------------------#

# Set the environment configuration.
service_key_file_location = '[SA_PATH]'

project_id = '[MY_PROJECT_ID]'

# Helper container to store results.
class DataContainer:
    def __init__(self):
        self.data = {}

    def callback(self, request_id, response, exception):
        if exception is not None:
            print('request_id: {}, exception: {}'.format(request_id, str(exception)))
            pass
        else:
            print(request_id)
            self.data[request_id] = response

# Helper function to build the Discovery Service config.
def get_service(api_name, api_version, scopes, key_file_location):
    """
    Get a service that communicates to a Google API.

    Args:
        api_name: The name of the API to connect to.
        api_version: The API version to connect to.
        scopes: A list auth scopes to authorize for the application.
        key_file_location: The path to a valid service account JSON key file.

    Returns:
        A service that is connected to the specified API.
    """
    credentials = ServiceAccountCredentials.from_json_keyfile_name(
        key_file_location, scopes=scopes)

    # Build the service object.
    service = build(api_name, api_version, credentials=credentials)

    return service

# Helper function to create a UUID for each request
def generated_uui():
    return str(uuid.uuid4())

def create_batch_request(callback):
    # For more info on supported regions
    # check: https://cloud.google.com/data-catalog/docs/concepts/regions

    region='us-datacatalog.googleapis.com'

    return BatchHttpRequest(batch_uri='https://{}/batch'.format(region), callback=callback)

container = DataContainer()

# Scope to set up the Discovery Service config.
scope = 'https://www.googleapis.com/auth/cloud-platform'

# Create service.
service = get_service(
    api_name='datacatalog',
    api_version='v1',
    scopes=[scope],
    key_file_location=service_key_file_location)

# Create the batch request config.
batch = create_batch_request(container.callback)

#-------------------------------------------------------------#
# 1. Start by fetching a list of entries using search call
#-------------------------------------------------------------#

# Create the search request body.
# This example searches for all BigQuery tables in a project.
search_request_body = {
  'query': 'type=TABLE system=BIGQUERY',
  'scope': {'includeProjectIds': [project_id]}
}

# Generated a unique ID for the request.
request_id = generated_uui()

# Add the request to the batch client.
batch.add(service.catalog().search(body=search_request_body), request_id=request_id)

# Execute the batch request.
batch.execute()

# Uncomment to verify the full response from search.
# print(container.data)

response = container.data[request_id]

results = response['results']

first_table = results[0]

# Verify that a first table is present.
print(first_table)

second_table = results[1]

# Verify that a second table is present
print(second_table)

#-------------------------------------------------------------------#
# 2. Send the batch request to attach tags over the entire result set
#-------------------------------------------------------------------#

# Create a new container
container = DataContainer()

# Create a new batch request
batch = create_batch_request(container.callback)

# Set the template name config
template_name = 'projects/[MY_PROJECT_ID]/locations/[MY-LOCATION]/tagTemplates/[MY-TEMPLATE-NAME]'

for result in results:
    # Generated a unique id for request.
    request_id = generated_uui()

    # Add the entry name as the tag parent.
    parent=result['relativeResourceName']

    # Create the tag request body.
    create_tag_request_body = {
      'template': template_name,
       # CHANGE for your template field values.
      'fields': {'etl_score': {'doubleValue': 0.5}}
    }

    # Add requests to the batch client.
    batch.add(service.projects().locations().
              entryGroups().entries().tags().
              create(body=create_tag_request_body,
                     parent=parent),
              request_id=request_id)

# Execute the batch request.

# Since the Batch Client works with regions
# If you receive [HttpError 400 errors]
# 1. Verify the region you used to create the Batch client
# 2. Verify the region where the Entry is located.
# 3. verify the region of the parent tag template used by the tag.

batch.execute()

# Uncomment to verify the full response from tag creation.
# print(container)