BigQuery 수집, 가상 테이블과 내보내기

BigQuery 커넥터는 BigQuery 테이블과 Foundry 데이터셋 사이에서 데이터를 전달합니다. 원천 탐색, 대량 가져오기, 증분 수집, 가상 테이블과 연산 위임을 일반 제공(GA)하며 Export tasks는 Sunset 단계입니다.

Data Connection의 +New Source에서 BigQuery를 선택합니다. 인증과 설정에는 Google Cloud IAM 서비스 계정이 필요합니다.

작업별 IAM 권한

인증과 네트워크

다음 호스트의 443 포트에 접근해야 합니다.

bigquery.googleapis.com
bigquerystorage.googleapis.com
storage.googleapis.com
www.googleapis.com

인증 구성에 따라 oauth2.googleapis.com, accounts.google.com 접근도 필요할 수 있습니다. GCP 기반 Foundry에서 GCP BigQuery로 직접 연결하면 관련 VPC Service Controls에서도 연결을 허용해야 하므로 필요한 경우 Palantir 지원을 통해 구성을 확인합니다.

소스 속성

Spark 호환 BIGNUMERIC 설정은 최대 precision 38, scale 18을 사용합니다. BigQuery 데이터가 이 범위를 넘으면 정보가 손실될 수 있습니다. precision이나 scale을 명시한 BIGNUMERIC이 이 상한을 넘는 경우는 지원하지 않습니다.

동기화와 자료형

Overview의 Explore and create syncs에서 테이블을 선택하고 Create sync for x datasets를 선택합니다. Edit syncs의 왼쪽 패널에서 대상 오른쪽 >로 상세 구성을 엽니다.

일반 데이터셋 수집은 Avro로 저장하며 BIGNUMERIC, TIME 열은 이 가져오기 경로에서 지원하지 않습니다. Foundry에서 BigQuery로 내보낼 때에는 MAPS, STRUCTS, ARRAYS를 제외한 열 유형을 지원합니다. 가상 테이블의 Spark 호환 설정과 일반 Avro 수집의 자료형 제한을 구분합니다.

전체 테이블

사용자 정의 SQL

질의는 select 또는 with로 시작해야 합니다. 예를 들어 SELECT * from table_name limit 100;을 사용할 수 있습니다. 결과가 20GB 미만이어야 하며, 더 큰 결과에는 임시 테이블 사용을 활성화합니다. 이 크기는 해당 질의 수집 경로의 조건이며 BigQuery 테이블 자체의 최대 크기를 뜻하지 않습니다.

증분 수집

일반 수집은 매번 조건에 맞는 모든 행을 가져옵니다. 증분은 이전 수집 위치를 유지하고 이후의 행만 읽습니다. 선택한 열은 엄격하게 단조 증가해야 합니다.

증분 열의 지원 후보 자료형은 다음과 같습니다. 실제 출력에는 앞서 설명한 수집 경로별 자료형 제한도 적용됩니다.

INT64, FLOAT64, NUMERIC, BIGNUMERIC, STRING,
TIMESTAMP, DATE, TIME, DATETIME

Column과 Initial value는 필수이며 Limit는 선택 항목입니다. 지원 자료형 열이 없으면 Column 드롭다운이 비어 있습니다.

가상으로 5TB 테이블의 id를 증분 열로 선택하고 초기값 -1, Limit 5천만 행을 지정하면 첫 실행은 id > -1의 행을 오름차순으로 가져옵니다. 이전 실행의 최대 ID가 19384004822이면 다음 실행은 그 값보다 큰 행부터 가져옵니다.

사용자 정의 질의를 증분 처리하려면 다음 형식으로 작성합니다. @value, @limit는 증분 실행의 값입니다.

SELECT * from table_name
where incremental_column_name > @value
order by incremental_column_name asc
limit @limit

임시 테이블

비표준 테이블이나 질의 결과가 20GB를 넘으면 임시 테이블을 사용합니다. 큰 질의 결과, 뷰와 대규모 증분 수집을 처리하는 경로입니다. 한 번의 수집 크기는 실행 디스크 여유 공간에 제한되며 Foundry 워커에서는 보통 600GB입니다. 더 큰 테이블은 증분으로 나눕니다.

Settings for temporary tables에서 다음 중 하나를 선택합니다.

• Automatically create dataset은 Palantir_temporary_tables라는 데이터셋을 자동 생성합니다. 프로젝트 수준의 BigQuery Data Editor가 필요합니다.
• Provide dataset to use는 Project ID와 Dataset name을 직접 지정합니다. 그 데이터셋의 BigQuery Data Editor가 필요합니다.

BigQuery 뷰를 가상 테이블로 등록할 때에도 임시 테이블을 활성화해야 합니다.

기존 Export task

BigQuery Export tasks는 Sunset 단계입니다. 기존 내보내기는 GCS를 중간 저장소로 사용하거나 BigQuery API를 직접 사용하는 두 경로가 있습니다. GCS 경로가 권장됩니다.

API 직접 경로는 수백만 행 수준에 적합하며 백만 행에 약 2분이 예상 성능의 예시입니다. 실제 시간 보장이 아닙니다. 수십억 행 규모에는 GCS 경로를 사용합니다.

프로젝트에서 소스를 오른쪽 클릭하고 Create Data Connection Task를 선택합니다. Source를 확인한 뒤 Input에 inputDataset, Output에 outputDataset을 지정합니다. 입력은 내보낼 데이터이고 출력은 작업 실행, 예약과 모니터링에 사용합니다. 패널 표시 이름은 YAML 식별자를 대신하지 않습니다.

Parquet으로 행이 저장된 Foundry 데이터셋만 내보낼 수 있습니다. 다음은 가상 대상에 대한 초기 SNAPSHOT 구성입니다.

type: magritte-bigquery-export-task
config:
    table:
      dataset: datasetInBigQuery
      table: tableInBigQuery
      project: projectId
    incrementalType: SNAPSHOT

대상 프로젝트가 소스 프로젝트와 같으면 project는 생략합니다. 첫 내보내기를 SNAPSHOT으로 수행한 뒤 증분 사용 시 incrementalType을 REQUIRE_INCREMENTAL로 바꿉니다. 증분 입력에는 행 추가만 있어야 하고 BigQuery 대상 스키마가 Foundry 입력 스키마와 일치해야 합니다.

GCS 경로

소스에 중간 GCS 버킷을 설정합니다. 이 버킷은 해당 커넥터의 임시 데이터 전용으로 사용하여 접근자를 최소화합니다. 큰 규모에 더 적합하며 BigQuery에 별도의 임시 테이블을 만들지 않습니다.

BigQuery API 경로

대상 옆에 datasetName.tableName_temp_$timestamp라는 임시 테이블을 만들고 먼저 데이터를 적재한 뒤 대상으로 옮깁니다. 임시 테이블에 별도 접근 제한을 적용하지 않습니다.

SNAPSHOT은 대상 테이블을 삭제하고 다시 만들기 때문에 별도 접근 제한과 공유 설정이 사라질 수 있습니다. REQUIRE_INCREMENTAL에서는 대상 공유를 유지할 수 있습니다. API 내보내기 대상에는 행 또는 열 수준 권한을 적용하지 않습니다.

Hive 테이블 파티션 데이터셋은 API 직접 내보내기를 지원하지 않으므로 GCS를 사용합니다.

외부 변환의 읽기와 쓰기

다음 예제는 google-cloud-bigquery와 JSON 서비스 계정 인증 정보를 사용합니다. 소스의 jsonCredentials 비밀 정보가 필요합니다. source RID, 입력/출력 RID, BigQuery 프로젝트, 데이터셋과 테이블을 실제 구성에 맞게 바꿉니다. 직접 코드 경로에서는 BigQuery Python client의 프로젝트 설정을 사용합니다.

예시 Shipments는 배송 건의 무게, 운송업체와 운송장 번호를 나타내는 테이블입니다.

운송업체와 무게로 조회

입력은 carrier_name, carrier_details 열을 가진 데이터셋입니다. 중복을 제거한 업체 목록과 무게 하한 10을 질의 매개변수로 전달합니다. 결과에는 shipment_id, carrier, weight, tracking_number가 있습니다.

from pandas import DataFrame
from transforms.api import lightweight, Output, Input, transform_pandas
from transforms.external.systems import ResolvedSource, external_systems, Source
from google.cloud import bigquery
import json

@lightweight
@external_systems(
    bigquery_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<output_dataset_rid>"),
    carriers_df=Input("<carriers_input_dataset_rid>") # dataframe with schema [carrier_name: String, carrier_details: String]
)
def read(bigquery_source: ResolvedSource, carriers_df: DataFrame) -> DataFrame:
    # Authenticate BigQuery client
    account_info = json.loads(bigquery_source.get_secret("jsonCredentials"))
    client = bigquery.Client.from_service_account_info(account_info)

    # Table reference
    bigquery_project = "<bigquery_project_id>"
    bigquery_dataset = "<bigquery_dataset_id>"
    bigquery_table = "<bigquery_table_id>"
    table_ref = f"{bigquery_project}.{bigquery_dataset}.{bigquery_table}"

    # Filter parameters
    carrier_list = carriers_df['carrier_name'].drop_duplicates().tolist()
    min_weight = 10

    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ArrayQueryParameter("carrier_list", "STRING", carrier_list),
            bigquery.ScalarQueryParameter("min_weight", "FLOAT64", min_weight)
        ]
    )

    query = f"""
        SELECT shipment_id, carrier, weight, tracking_number
        FROM `{table_ref}`
        WHERE carrier IN UNNEST(@carrier_list) AND weight >= @min_weight
    """

    query_job = client.query(query, job_config=job_config)
    return query_job.to_dataframe()

운송장 번호 갱신

입력은 shipment_id: Long, tracking_number: String입니다. 작업은 고정 이름 temp_update_tracking의 기존 임시 테이블을 삭제하고 입력을 적재한 뒤, 일치하는 배송 건의 운송장 번호를 MERGE로 갱신합니다. 그 후 임시 테이블을 삭제하고 대상 행을 다시 조회하여 반환합니다.

이 이름은 작업 전용 임시 테이블이어야 하며 같은 이름을 사용하는 작업을 동시에 실행하지 않습니다. 대상 테이블의 외부 데이터가 변경됩니다. 중간 오류가 발생하면 뒤의 정리 단계까지 진행하지 못할 수 있습니다.

from pandas import DataFrame
from transforms.api import lightweight, Output, Input, transform_pandas
from transforms.external.systems import ResolvedSource, external_systems, Source
from google.cloud import bigquery
import json

@lightweight
@external_systems(
    bigquery_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<output_dataset_rid>"),
    shipments_df=Input("<shipments_input_dataset_rid>"), # DataFrame with schema [shipment_id: Long, tracking_number: String]
)
def update_tracking_numbers(bigquery_source: ResolvedSource, shipments_df: DataFrame) -> DataFrame:
    # Authenticate BigQuery client
    account_info = json.loads(bigquery_source.get_secret("jsonCredentials"))
    client = bigquery.Client.from_service_account_info(account_info)

    # Table reference
    bigquery_project = "<bigquery_project_id>"
    bigquery_dataset = "<bigquery_dataset_id>"
    bigquery_table = "<bigquery_table_id>"
    table_ref = f"{bigquery_project}.{bigquery_dataset}.{bigquery_table}"

    # 1. Clear temporary table if it exists
    temp_table_id = f"{bigquery_project}.{bigquery_dataset}.temp_update_tracking"
    client.delete_table(temp_table_id, not_found_ok=True)


    # 2. Upload shipments_df to a temporary table
    load_job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("shipment_id", "INT64"),
            bigquery.SchemaField("tracking_number", "STRING"),
        ]
    )
    job = client.load_table_from_dataframe(shipments_df, temp_table_id, job_config=load_job_config)
    job.result()  # Wait for load to finish

    # 3. MERGE to update tracking_number in main table
    merge_query = f"""
    MERGE `{table_ref}` T
    USING `{temp_table_id}` S
    ON T.shipment_id = S.shipment_id
    WHEN MATCHED THEN
      UPDATE SET tracking_number = S.tracking_number
    """

    client.query(merge_query).result()

    # 3. (Optional) Clean up temporary table
    client.delete_table(temp_table_id, not_found_ok=True)

    # Optionally, return the updated rows as a DataFrame for confirmation
    updated_ids = shipments_df["shipment_id"].tolist()
    query = f"""
        SELECT shipment_id, tracking_number
        FROM `{table_ref}`
        WHERE shipment_id IN UNNEST(@updated_ids)
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ArrayQueryParameter("updated_ids", "INT64", [int(x) for x in updated_ids])
        ]
    )
    updated_rows = client.query(query, job_config=job_config).to_dataframe()
    return updated_rows

가상 테이블과 연산 위임

대량 등록과 자동 등록을 일반 제공합니다. Code Repositories와 Pipeline Builder에서 테이블, 뷰와 구체화 뷰를 입력으로 사용할 수 있고 테이블을 출력으로 사용할 수 있습니다. 증분 파이프라인은 APPEND만 지원합니다.

증분에는 적절한 보존 기간의 Time Travel이 필요합니다. Change History를 사용하고 Python Transforms의 current, added 모드에서 _CHANGE_TYPE, _CHANGE_TIMESTAMP를 제공합니다. 이 연결 경로의 증분 처리는 UPDATE나 DELETE 변경을 반영하는 방식이 아니므로 원천도 추가 전용 테이블이어야 합니다.

가상 테이블의 조건은 다음과 같습니다.

• Foundry 워커 소스를 사용합니다. 에이전트 워커는 지원하지 않습니다.
• 서비스 계정 키 또는 OIDC를 사용합니다. 인스턴스 인증은 지원하지 않습니다.
• 필요한 양방향 연결과 허용 목록을 구성합니다.
• 뷰에는 임시 테이블과 그 생성/사용 권한이 필요합니다.
• 매개변수가 없는 BIGNUMERIC을 Spark로 읽으려면 Spark 호환 설정을 켭니다.

가상 테이블 입력은 BigQuery Spark 커넥터의 조건절 위임을 사용합니다. 같은 소스의 가상 테이블을 입력과 출력으로 구성한 파이프라인은 Python Transforms에서 연산 전체를 BigQuery로 위임할 수 있습니다.

문제 확인

데이터셋 위치 오류

Not found: Dataset <dataset> was not found in location <location>이면 질의 입력과 임시 테이블 데이터셋이 같은 리전인지 확인합니다. 임시 테이블을 사용하면 그 데이터셋 위치가 질의 실행 위치를 결정합니다. 자동 생성한 경우 Google Cloud에서 Palantir_temporary_tables의 위치를 확인합니다.

gRPC 최대 메시지 크기

큰 JSON 열 등으로 gRPC message exceeds maximum size가 발생하면 Connection Settings, Connection Details의 More options에서 gRPC Settings를 열어 Maximum inbound message size를 늘립니다. 호출 하나가 여러 행을 가져오므로 가장 큰 행 하나의 크기만으로는 부족할 수 있습니다.

뷰 미리보기 실패

뷰의 스키마는 생성 시 질의 결과로 결정되며 기반 테이블의 이후 구조 변경이 자동 반영되지 않을 수 있습니다. 뷰와 기반 테이블의 열 존재 여부와 자료형을 비교합니다. 차이가 있으면 CREATE OR REPLACE VIEW 또는 ALTER VIEW로 뷰를 갱신하거나 다시 만듭니다.