애플리케이션이 만드는 각 HTTP 연결에는 어느 정도의 오버헤드가 필요합니다. Data Catalog API 요청은 여러 API 호출을 단일 HTTP 요청으로 결합할 수 있도록 일괄 처리를 지원합니다. 수행할 작은 요청이 여러 개 있고 HTTP 요청 오버헤드를 최소화하기 위해서는 HTTP 일괄 처리를 사용할 수 있습니다. 일괄 처리로 오버헤드가 줄어들지만 API 할당량 계산에서 일괄 처리 내의 요청은 여전히 여러 요청으로 계산됩니다.
Google Cloud에서 HTTP 배치를 사용하는 방법에 관한 일반 문서는 Google API Python 클라이언트 문서를 참조하세요.
Python에서 HTTP 일괄 요청 만들기
일괄 요청을 사용하여 Data Catalog의 항목을 만들거나 조작하려면 먼저 catalog.search() 또는 entries.lookup()을 사용하여 변경하려는 항목을 검색해야 합니다.
이제 다음 단계에 따라 Google Python API을 사용하여 HTTP 일괄 요청을 빌드합니다.
new_batch_http_request()를 호출하거나BatchHttpRequest()생성자를 사용하여BatchHttpRequest객체를 만듭니다. 각 요청에 대한 응답으로 호출되는 콜백을 전달할 수 있습니다.- 실행하려는 각 요청에 대해
BatchHttpRequest객체에서add()를 호출합니다.BatchHttpRequest객체를 만들 때 콜백을 전달한 경우 각add()에는 콜백에 전달할 매개변수가 포함될 수 있습니다. - 요청을 추가한 후
BatchHttpRequest객체에서execute()를 호출하여 실행합니다.execute()함수는 모든 콜백이 호출될 때까지 차단됩니다.
BatchHttpRequest의 요청은 병렬로 실행될 수 있으며, 실행 순서가 보장되지 않습니다. 즉, 동일 배치에 있는 요청이 서로 종속되지 않습니다. 예를 들어 EntryGroup을 만들고 여기에 속하는 Entry를 동일한 요청으로 만들지 않아야 합니다. EntryGroup 만들기가 수행되기 전 Entry 만들기가 수행되어 결국 실행이 실패할 수 있기 때문입니다.
리전 엔드포인트가 있는 일괄 요청
Data Catalog 리전 API 엔드포인트에 HTTP 일괄 요청을 사용할 때 모든 API 요청은 동일한 리전에 속해야 합니다. 배치를 실행할 때는 올바른 리전 엔드포인트를 호출해야 합니다. 예를 들어 리소스가 us-central1에 있으면 https://us-central1-datacatalog.googleapis.com/batch를 호출합니다.
리전 독립 API
리전 독립 API(예: catalog.lookup() 및 entries.search())는 서로 그룹으로 묶을 수 있지만, 리전 종속 API와 그룹으로 묶지 않아야 합니다. 리전 독립 API의 경우 https://datacatalog.googleapis.com/batch 엔드포인트를 사용합니다.
예
이 샘플 Python 애플리케이션은 HTTP 일괄 요청을 사용하여 Data Catalog API로 태그 템플릿으로부터 여러 태그를 만드는 방법을 보여줍니다.
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)