Cloud Storage CSV Files to BigQuery 템플릿

Cloud Storage CSV files to BigQuery 파이프라인은 Cloud Storage에 저장된 CSV 파일에서 데이터를 읽고 결과를 BigQuery 테이블에 추가할 수 있게 해주는 일괄 파이프라인입니다. CSV 파일은 압축되지 않거나 Compression Enum SDK 페이지에 나와 있는 형식으로 압축할 수 있습니다.

파이프라인 요구사항

이 템플릿을 사용하려면 파이프라인이 다음 요구사항을 충족해야 합니다.

BigQuery 스키마 JSON 파일

BigQuery 스키마를 설명하는 JSON 파일을 만듭니다. 스키마에 BigQuery Schema라는 최상위 JSON 배열이 있고 해당 콘텐츠는 {"name": "COLUMN_NAME", "type": "DATA_TYPE"} 패턴을 따라야 합니다.

Cloud Storage CSV files to BigQuery 일괄 템플릿은 데이터를 대상 BigQuery 테이블에서 STRUCT(레코드) 필드로 가져올 수 없습니다.

다음 JSON은 예시 BigQuery 스키마를 설명합니다.

{
  "BigQuery Schema": [
    {
      "name": "location",
      "type": "STRING"
    },
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "STRING"
    },
    {
      "name": "color",
      "type": "STRING"
    },
    {
      "name": "coffee",
      "type": "STRING"
    }
  ]
}

오류 테이블 스키마

CSV 파일에서 거부된 레코드를 저장하는 BigQuery 테이블은 여기에 정의된 테이블 스키마와 일치해야 합니다.

{
  "BigQuery Schema": [
    {
      "name": "RawContent",
      "type": "STRING"
    },
    {
      "name": "ErrorMsg",
      "type": "STRING"
    }
  ]
}

템플릿 매개변수

필수 매개변수

  • inputFilePattern: 처리할 텍스트가 포함된 CSV 파일에 대한 Cloud Storage 경로입니다. (예: gs://your-bucket/path/*.csv)
  • schemaJSONPath: BigQuery 스키마를 정의하는 JSON 파일에 대한 Cloud Storage 경로입니다.
  • outputTable: 처리된 데이터가 저장된 BigQuery 테이블의 이름입니다. 기존 BigQuery 테이블을 다시 사용하면 데이터가 대상 테이블에 추가됩니다.
  • bigQueryLoadingTemporaryDirectory: BigQuery 로드 프로세스 중에 사용할 임시 디렉터리입니다. (예: gs://your-bucket/your-files/temp_dir)
  • badRecordsOutputTable: CSV 파일을 처리할 때 거부된 데이터를 저장하는 데 사용할 BigQuery 테이블의 이름입니다. 기존 BigQuery 테이블을 다시 사용하면 데이터가 대상 테이블에 추가됩니다. 이 테이블의 스키마는 오류 테이블 스키마(https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-csv-to-bigquery#GcsCSVToBigQueryBadRecordsSchema)와 일치해야 합니다.
  • delimiter: CSV 파일이 사용하는 열 구분 기호입니다. (예: ,)
  • csvFormat: Apache Commons CSV 형식에 따른 CSV 형식입니다. 기본값은 Default입니다.

선택적 매개변수

  • containsHeaders: 헤더가 CSV 파일에 포함되어 있는지 여부입니다. 기본값은 false입니다.
  • csvFileEncoding: CSV 파일 문자 인코딩 형식입니다. 허용되는 값은 US-ASCII, ISO-8859-1, UTF-8, UTF-16입니다. 기본값은 UTF-8입니다.

템플릿 실행

콘솔

  1. Dataflow 템플릿에서 작업 만들기 페이지로 이동합니다.
  2. 템플릿에서 작업 만들기로 이동
  3. 작업 이름 필드에 고유한 작업 이름을 입력합니다.
  4. (선택사항): 리전 엔드포인트의 드롭다운 메뉴에서 값을 선택합니다. 기본 리전은 us-central1입니다.

    Dataflow 작업을 실행할 수 있는 리전 목록은 Dataflow 위치를 참조하세요.

  5. Dataflow 템플릿 드롭다운 메뉴에서 the CSV files on Cloud Storage to BigQuery (Batch) template을 선택합니다.
  6. 제공된 매개변수 필드에 매개변수 값을 입력합니다.
  7. 작업 실행을 클릭합니다.

gcloud

셸 또는 터미널에서 템플릿을 실행합니다.

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/GCS_CSV_to_BigQuery \
    --region REGION_NAME \
    --parameters \
inputFilePattern=PATH_TO_CSV_DATA,\
schemaJSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
outputTable=BIGQUERY_DESTINATION_TABLE,\
badRecordsOutputTable=BIGQUERY_BAD_RECORDS_TABLE,\
csvFormat=CSV_FORMAT,\
delimiter=DELIMITER,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
containsHeaders=CONTAINS_HEADERS,\
csvFileEncoding=CSV_FILE_ENCODING

다음을 바꿉니다.

  • PROJECT_ID: Dataflow 작업을 실행하려는 Google Cloud 프로젝트 ID
  • JOB_NAME: 선택한 고유한 작업 이름
  • VERSION: 사용할 템플릿 버전

    다음 값을 사용할 수 있습니다.

  • REGION_NAME: Dataflow 작업을 배포할 리전(예: us-central1)
  • PATH_TO_CSV_DATA: CSV 파일에 대한 Cloud Storage 경로
  • PATH_TO_BIGQUERY_SCHEMA_JSON: 스키마 정의가 포함된 JSON 파일의 Cloud Storage 경로
  • BIGQUERY_DESTINATION_TABLE: BigQuery 대상 테이블 이름
  • BIGQUERY_BAD_RECORDS_TABLE: BigQuery 불량 레코드 테이블 이름
  • PATH_TO_TEMP_DIR_ON_GCS: 임시 디렉터리의 Cloud Storage 경로
  • DELIMITER: CSV 파일 구분 기호
  • CSV_FORMAT: 레코드 파싱을 위한 CSV 형식 사양
  • CONTAINS_HEADERS: CSV 파일에 헤더가 포함되어 있는지 여부
  • CSV_FILE_ENCODING: CSV 파일에서 인코딩

API

REST API를 사용하여 템플릿을 실행하려면 HTTP POST 요청을 전송합니다. API 및 승인 범위에 대한 자세한 내용은 projects.templates.launch를 참조하세요.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/GCS_CSV_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern":"PATH_TO_CSV_DATA",
       "schemaJSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "outputTable":"BIGQUERY_DESTINATION_TABLE",
       "badRecordsOutputTable":"BIGQUERY_BAD_RECORDS_TABLE",
       "csvFormat":"CSV_FORMAT",
       "delimiter":"DELIMITER",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
       "containsHeaders": "CONTAINS_HEADERS",
       "csvFileEncoding": "CSV_FILE_ENCODING"
   },
   "environment": { "zone": "us-central1-f" }
}

다음을 바꿉니다.

  • PROJECT_ID: Dataflow 작업을 실행하려는 Google Cloud 프로젝트 ID
  • JOB_NAME: 선택한 고유한 작업 이름
  • VERSION: 사용할 템플릿 버전

    다음 값을 사용할 수 있습니다.

  • LOCATION: Dataflow 작업을 배포할 리전(예: us-central1)
  • PATH_TO_CSV_DATA: CSV 파일에 대한 Cloud Storage 경로
  • PATH_TO_BIGQUERY_SCHEMA_JSON: 스키마 정의가 포함된 JSON 파일의 Cloud Storage 경로
  • BIGQUERY_DESTINATION_TABLE: BigQuery 대상 테이블 이름
  • BIGQUERY_BAD_RECORDS_TABLE: BigQuery 불량 레코드 테이블 이름
  • PATH_TO_TEMP_DIR_ON_GCS: 임시 디렉터리의 Cloud Storage 경로
  • DELIMITER: CSV 파일 구분 기호
  • CSV_FORMAT: 레코드 파싱을 위한 CSV 형식 사양
  • CONTAINS_HEADERS: CSV 파일에 헤더가 포함되어 있는지 여부
  • CSV_FILE_ENCODING: CSV 파일에서 인코딩

다음 단계