BigQuery to Elasticsearch 템플릿

BigQuery to Elasticsearch 템플릿은 BigQuery 테이블의 데이터를 Elasticsearch에 문서로 수집하는 일괄 파이프라인입니다. 이 템플릿에서는 전체 테이블을 읽거나 제공된 쿼리를 사용하여 특정 레코드를 읽을 수 있습니다.

파이프라인 요구사항

소스 BigQuery 테이블이 있어야 합니다.
Google Cloud 인스턴스 또는 Elasticsearch 버전 7.0 이상을 사용하는 Elastic Cloud의 Elasticsearch 호스트. Dataflow 작업자 머신에서 액세스할 수 있어야 합니다.

템플릿 매개변수

필수 매개변수

connectionUrl: https://hostname:[port] 형식의 Elasticsearch URL입니다. Elastic Cloud를 사용하는 경우 CloudID를 지정합니다. (예: https://elasticsearch-host:9200)
apiKey: 인증에 사용할 Base64로 인코딩된 API 키입니다.
색인: my-index.와 같이 요청이 실행되는 Elasticsearch 색인입니다(예: my-index).

선택적 매개변수

inputTableSpec: 읽을 BigQuery 테이블입니다. 형식은 projectId:datasetId.tablename입니다. inputTableSpec을 지정하는 경우 템플릿은 BigQuery Storage Read API(https://cloud.google.com/bigquery/docs/reference/storage)를 사용하여 BigQuery 스토리지에서 직접 데이터를 읽습니다. Storage Read API의 제한사항에 대한 자세한 내용은 https://cloud.google.com/bigquery/docs/reference/storage#limitations를 참조하세요. inputTableSpec 또는 query를 지정해야 합니다. 두 매개변수를 모두 설정하면 템플릿에서 query 매개변수를 사용합니다. (예: bigquery-project:dataset.input_table)
outputDeadletterTable: 출력 테이블에 도달하지 못한 메시지의 BigQuery 테이블로 <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE> 형식입니다. 테이블이 없으면 파이프라인 실행 중에 생성됩니다. 지정하지 않으면 <outputTableSpec>_error_records이 사용됩니다. (예: your-project-id:your-dataset.your-table-name)
query: BigQuery에서 데이터를 읽는 데 사용할 SQL 쿼리입니다. BigQuery 데이터 세트가 Dataflow 작업과 다른 프로젝트에 있는 경우 SQL 쿼리에 전체 데이터 세트 이름을 지정합니다(예: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>). useLegacySql이 true가 아닌 한 기본적으로 query 매개변수는 GoogleSQL(https://cloud.google.com/bigquery/docs/introduction-sql)을 사용합니다. inputTableSpec 또는 query를 지정해야 합니다. 두 매개변수를 모두 설정하면 템플릿에서 query 매개변수를 사용합니다. (예: sampledb.sample_table에서 * 선택).
useLegacySql: legacy SQL을 사용하려면 true로 설정합니다. 이 매개변수는 query 매개변수를 사용할 때만 적용됩니다. 기본값은 false입니다.
queryLocation: 기본 테이블 권한 없이 승인된 뷰에서 읽을 때 필요합니다. (예: US)
elasticsearchUsername: 인증할 Elasticsearch 사용자 이름입니다. 지정하면 'apiKey' 값이 무시됩니다.
elasticsearchPassword: 인증할 Elasticsearch 비밀번호입니다. 지정하면 'apiKey' 값이 무시됩니다.
batchSize: 문서 수의 배치 크기입니다. 기본값은 1000입니다.
batchSizeBytes: 바이트 수의 배치 크기입니다. 기본값은 5242880(5mb)입니다.
maxRetryAttempts: 최대 재시도 시도 횟수입니다. 0보다 커야 합니다. 기본값은 재시도 없음입니다.
maxRetryDuration: 최대 재시도 시간(밀리초)입니다. 0보다 커야 합니다. 기본값은 재시도 없음입니다.
propertyAsIndex: 색인이 생성되는 문서의 속성으로, 값이 일괄 요청에서 문서에 포함할 _index 메타데이터를 지정합니다. _index UDF보다 우선 적용됩니다. 기본값은 없음입니다.
javaScriptIndexFnGcsPath: 일괄 요청에서 문서에 포함할 _index 메타데이터를 지정하는 함수의 JavaScript UDF 소스에 대한 Cloud Storage 경로입니다. 기본값은 없음입니다.
javaScriptIndexFnName: 일괄 요청에서 문서에 포함할 _index 메타데이터를 지정하는 UDF JavaScript 함수의 이름입니다. 기본값은 없음입니다.
propertyAsId: 색인이 생성되는 문서의 속성으로, 값이 일괄 요청에서 문서에 포함할 _id 메타데이터를 지정합니다. _id UDF보다 우선 적용됩니다. 기본값은 없음입니다.
javaScriptIdFnGcsPath: 일괄 요청에서 문서에 포함할 _id 메타데이터를 지정하는 함수의 JavaScript UDF 소스에 대한 Cloud Storage 경로입니다. 기본값은 없음입니다.
javaScriptIdFnName: 일괄 요청에서 문서에 포함할 _id 메타데이터를 지정하는 UDF JavaScript 함수의 이름입니다. 기본값은 없음입니다.
javaScriptTypeFnGcsPath: 일괄 요청에서 문서에 포함할 _type 메타데이터를 지정하는 함수의 JavaScript UDF 소스에 대한 Cloud Storage 경로입니다. 기본값: none.
javaScriptTypeFnName: 일괄 요청에서 문서에 포함할 _type 메타데이터를 지정하는 UDF JavaScript 함수의 이름입니다. 기본값은 없음입니다.
javaScriptIsDeleteFnGcsPath: 문서를 삽입하거나 업데이트하는 대신 삭제할지 여부를 결정하는 함수의 JavaScript UDF 소스에 대한 Cloud Storage 경로입니다. 이 함수는 true 또는 false의 문자열 값을 반환합니다. 기본값은 없음입니다.
javaScriptIsDeleteFnName: 문서를 삽입 또는 업데이트하는 대신 삭제할지 여부를 결정하는 UDF JavaScript 함수의 이름입니다. 이 함수는 true 또는 false의 문자열 값을 반환합니다. 기본값은 없음입니다.
usePartialUpdate: Elasticsearch 요청에서 부분 업데이트(만들기 또는 색인 생성 대신 업데이트, 부분 문서 허용)를 사용할지 여부입니다. 기본값은 false입니다.
bulkInsertMethod: Elasticsearch 일괄 요청에서 INDEX(색인, 업데이트 허용) 또는 CREATE(만들기, 중복 _id 오류)를 사용할지 여부입니다. 기본값은 CREATE입니다.
trustSelfSignedCerts: 자체 서명 인증서를 신뢰할지 여부입니다. 설치된 Elasticsearch 인스턴스에 자체 서명 인증서가 있을 수 있습니다. SSL 인증서 유효성 검사를 우회하려면 이를 True로 사용 설정합니다. 기본값은 false입니다.
disableCertificateValidation: 'true'인 경우 자체 서명 SSL 인증서를 신뢰합니다. Elasticsearch 인스턴스에는 자체 서명 인증서가 있을 수 있습니다. 인증서 유효성 검사를 우회하려면 이 매개변수를 'true'로 설정합니다. 기본값: false.
apiKeyKMSEncryptionKey: API 키를 복호화하는 Cloud KMS 키입니다. apiKeySource가 KMS로 설정되면 이 매개변수를 제공해야 합니다. 이 매개변수가 제공되면 apiKey 문자열을 암호화하여 전달해야 합니다. KMS API 암호화 엔드포인트를 사용하여 매개변수를 암호화합니다. 키는 projects/{gcp_project}/locations/{key_region}/keyRings/{key_ring}/cryptoKeys/{kms_key_name} 형식이어야 합니다. https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt를 참조하세요(예: projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name).
apiKeySecretId: apiKey의 Secret Manager 보안 비밀 ID입니다. apiKeySource가 SECRET_MANAGER로 설정되면 이 매개변수를 제공해야 합니다. projects/{project}/secrets/{secret}/versions/{secret_version} 형식이어야 합니다. (예: projects/your-project-id/secrets/your-secret/versions/your-secret-version).
apiKeySource: API 키의 소스입니다. PLAINTEXT, KMS 또는 SECRET_Manager 중 하나입니다. Secret Manager 또는 KMS를 사용하는 경우 이 매개변수를 제공해야 합니다. apiKeySource가 KMS로 설정되면 apiKeyKMSEncryptionKey 및 암호화된 apiKey를 제공해야 합니다. apiKeySource가 SECRET_MANAGER로 설정되면 apiKeySecretId를 제공해야 합니다. apiKeySource가 PLAINTEXT로 설정되면 apiKey를 제공해야 합니다. 기본값은 PLAINTEXT입니다.
javascriptTextTransformGcsPath : 사용할 JavaScript 사용자 정의 함수(UDF)를 정의하는 .js 파일의 Cloud Storage URI입니다. (예: gs://my-bucket/my-udfs/my_file.js).
javascriptTextTransformFunctionName : 사용할 JavaScript 사용자 정의 함수(UDF)의 이름입니다. 예를 들어 JavaScript 함수가 myTransform(inJson) { /*...do stuff...*/ }이면 함수 이름은 myTransform입니다. 샘플 JavaScript UDF는 UDF 예시(https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)를 참조하세요.

사용자 정의 함수

이 템플릿은 아래 설명된 파이프라인의 여러 포인트에서 사용자 정의 함수(UDF)를 지원합니다. 자세한 내용은 Dataflow 템플릿에 대한 사용자 정의 함수 만들기를 참조하세요.

색인 함수

문서가 속한 색인을 반환합니다.

템플릿 매개변수:

javaScriptIndexFnGcsPath: JavaScript 파일의 Cloud Storage URI입니다.
javaScriptIndexFnName: JavaScript 함수의 이름입니다.

함수 사양:

입력: JSON 문자열로 직렬화된 Elasticsearch 문서입니다.
출력: 문서의 _index 메타데이터 필드 값입니다.

문서 ID 함수

문서 ID를 반환합니다.

템플릿 매개변수:

javaScriptIdFnGcsPath: JavaScript 파일의 Cloud Storage URI입니다.
javaScriptIdFnName: JavaScript 함수의 이름입니다.

함수 사양:

입력: JSON 문자열로 직렬화된 Elasticsearch 문서입니다.
출력: 문서의 _id 메타데이터 필드 값입니다.

문서 삭제 함수

문서 삭제 여부를 지정합니다. 이 함수를 사용하려면 대량 삽입 모드를 INDEX로 설정하고 문서 ID 함수를 제공합니다.

템플릿 매개변수:

javaScriptIsDeleteFnGcsPath: JavaScript 파일의 Cloud Storage URI입니다.
javaScriptIsDeleteFnName: JavaScript 함수의 이름입니다.

함수 사양:

입력: JSON 문자열로 직렬화된 Elasticsearch 문서입니다.
출력: 문서를 삭제하려면 문자열을 "true"로 반환하고 문서를 삽입/업데이트하려면 "false"로 반환합니다.

매핑 유형 함수

문서의 매핑 유형을 반환합니다.

템플릿 매개변수:

javaScriptTypeFnGcsPath: JavaScript 파일의 Cloud Storage URI입니다.
javaScriptTypeFnName: JavaScript 함수의 이름입니다.

함수 사양:

입력: JSON 문자열로 직렬화된 Elasticsearch 문서입니다.
출력: 문서의 _type 메타데이터 필드 값입니다.

템플릿 실행

콘솔

Dataflow 템플릿에서 작업 만들기 페이지로 이동합니다.

템플릿에서 작업 만들기로 이동

작업 이름 필드에 고유한 작업 이름을 입력합니다.
(선택사항): 리전 엔드포인트의 드롭다운 메뉴에서 값을 선택합니다. 기본 리전은 us-central1입니다.
Dataflow 작업을 실행할 수 있는 리전 목록은 Dataflow 위치를 참조하세요.
Dataflow 템플릿 드롭다운 메뉴에서 the BigQuery to Elasticsearch template을 선택합니다.
제공된 매개변수 필드에 매개변수 값을 입력합니다.
작업 실행을 클릭합니다.

gcloud

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

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/BigQuery_to_Elasticsearch \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

다음을 바꿉니다.

PROJECT_ID: Dataflow 작업을 실행하려는 Google Cloud 프로젝트 ID
JOB_NAME: 선택한 고유한 작업 이름
REGION_NAME: Dataflow 작업을 배포할 리전(예: us-central1)
VERSION: 사용할 템플릿 버전
다음 값을 사용할 수 있습니다.
- latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates-REGION_NAME/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
- 버전 이름(예: 2023-09-12-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates-REGION_NAME/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
INPUT_TABLE_SPEC: BigQuery 테이블 이름입니다.
CONNECTION_URL: Elasticsearch URL입니다.
APIKEY: 인증을 위한 base64 인코딩 API 키입니다.
INDEX: Elasticsearch 색인입니다.

API

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/BigQuery_to_Elasticsearch",
   }
}

다음을 바꿉니다.

PROJECT_ID: Dataflow 작업을 실행하려는 Google Cloud 프로젝트 ID
JOB_NAME: 선택한 고유한 작업 이름
LOCATION: Dataflow 작업을 배포할 리전(예: us-central1)
VERSION: 사용할 템플릿 버전
다음 값을 사용할 수 있습니다.
- latest: 버킷의 날짜가 지정되지 않은 상위 폴더(gs://dataflow-templates-REGION_NAME/latest/)에서 사용할 수 있는 최신 버전의 템플릿을 사용합니다.
- 버전 이름(예: 2023-09-12-00_RC00): 버킷의 날짜가 지정된 해당 상위 폴더(gs://dataflow-templates-REGION_NAME/)에 중첩되어 있는 특정 버전의 템플릿을 사용합니다.
주의: 최신 버전의 템플릿이 브레이킹 체인지로 업데이트될 수 있습니다. 프로덕션 환경에서는 이러한 브레이킹 체인지가 프로덕션 워크플로에 영향을 미치지 않도록 최신 날짜가 지정된 상위 폴더에 보관된 템플릿을 사용해야 합니다.
INPUT_TABLE_SPEC: BigQuery 테이블 이름입니다.
CONNECTION_URL: Elasticsearch URL입니다.
APIKEY: 인증을 위한 base64 인코딩 API 키입니다.
INDEX: Elasticsearch 색인입니다.

템플릿 소스 코드

Java

/*
 * Copyright (C) 2021 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.elasticsearch.templates;

import com.google.cloud.teleport.metadata.MultiTemplate;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.BigQueryToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.ReadBigQueryTableRows;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.TableRowToJsonFn;
import com.google.cloud.teleport.v2.transforms.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.v2.transforms.PythonExternalTextTransformer;
import com.google.common.base.Strings;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;

/**
 * The {@link BigQueryToElasticsearch} pipeline exports data from a BigQuery table to Elasticsearch.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/README_BigQuery_to_Elasticsearch.md">README</a>
 * for instructions on how to use or modify this template.
 */
@MultiTemplate({
  @Template(
      name = "BigQuery_to_Elasticsearch",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch",
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "pythonExternalTextTransformGcsPath",
        "pythonExternalTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      }),
  @Template(
      name = "BigQuery_to_Elasticsearch_Xlang",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch with Python UDFs",
      type = Template.TemplateType.XLANG,
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "javascriptTextTransformGcsPath",
        "javascriptTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch-xlang",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      })
})
public class BigQueryToElasticsearch {
  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    BigQueryToElasticsearchOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigQueryToElasticsearchOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(BigQueryToElasticsearchOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);
    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Create json string from Table Row.
     *        3) Write records to Elasticsearch.
     */

    boolean useJavascriptUdf = !Strings.isNullOrEmpty(options.getJavascriptTextTransformGcsPath());
    boolean usePythonUdf = !Strings.isNullOrEmpty(options.getPythonExternalTextTransformGcsPath());
    if (useJavascriptUdf && usePythonUdf) {
      throw new IllegalArgumentException(
          "Either javascript or Python gcs path must be provided, but not both.");
    }

    /*
     * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
     */
    PCollection<String> readJsonDocuments =
        pipeline
            .apply(
                "ReadFromBigQuery",
                ReadBigQueryTableRows.newBuilder()
                    .setOptions(options.as(BigQueryToElasticsearchOptions.class))
                    .build())

            /*
             * Step #2: Convert table rows to JSON documents.
             */
            .apply("TableRowsToJsonDocument", ParDo.of(new TableRowToJsonFn()));

    /*
     * Step #3: Apply UDF functions (if specified)
     */
    PCollection<String> udfOut;
    if (usePythonUdf) {
      udfOut =
          readJsonDocuments
              .apply(
                  "MapToRecord",
                  PythonExternalTextTransformer.FailsafeRowPythonExternalUdf
                      .stringMappingFunction())
              .setRowSchema(PythonExternalTextTransformer.FailsafeRowPythonExternalUdf.ROW_SCHEMA)
              .apply(
                  "InvokeUDF",
                  PythonExternalTextTransformer.FailsafePythonExternalUdf.newBuilder()
                      .setFileSystemPath(options.getPythonExternalTextTransformGcsPath())
                      .setFunctionName(options.getPythonExternalTextTransformFunctionName())
                      .build())
              .apply(
                  "MapToStringElements",
                  ParDo.of(new PythonExternalTextTransformer.RowToStringElementFn()));
    } else {
      udfOut =
          readJsonDocuments.apply(
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }

    /*
     * Step #4: Write converted records to Elasticsearch
     */
    udfOut.apply(
        "WriteToElasticsearch",
        WriteToElasticsearch.newBuilder()
            .setUserAgent("dataflow-bigquery-to-elasticsearch-template/v2")
            .setOptions(options.as(BigQueryToElasticsearchOptions.class))
            .build());

    return pipeline.run();
  }
}

다음 단계

Dataflow 템플릿 알아보기
Google 제공 템플릿 목록 참조