Template PostgreSQL ke BigQuery

Template PostgreSQL ke BigQuery adalah pipeline batch yang menyalin data dari tabel PostgreSQL ke dalam tabel BigQuery yang sudah ada. Pipeline ini menggunakan JDBC untuk terhubung ke PostgreSQL. Sebagai lapisan perlindungan tambahan, Anda juga dapat meneruskan kunci Cloud KMS bersama dengan nama pengguna, sandi, dan parameter string koneksi yang dienkode menggunakan Base64 yang dienkripsi dengan kunci Cloud KMS. Untuk mengetahui informasi selengkapnya tentang enkripsi nama pengguna, sandi, dan parameter string koneksi Anda, lihat endpoint enkripsi Cloud KMS API.

Persyaratan pipeline

• Tabel BigQuery harus ada sebelum eksekusi pipeline.
• Tabel BigQuery harus memiliki skema yang kompatibel.
• Database relasional harus dapat diakses dari subnet tempat Dataflow berjalan.

Parameter template

Parameter	Deskripsi
connectionURL	String URL koneksi JDBC. Misalnya, jdbc:postgresql://some-host:5432/sampledb. Anda dapat meneruskan nilai ini sebagai string yang dienkripsi dengan kunci Cloud KMS lalu berenkode Base64. Menghapus karakter spasi kosong dari string berenkode Base64.
outputTable	Lokasi tabel output BigQuery, dalam format <my-project>:<my-dataset>.<my-table>.
bigQueryLoadingTemporaryDirectory	Direktori sementara untuk proses pemuatan BigQuery. Misalnya, gs://<my-bucket>/my-files/temp_dir.
query	Kueri yang akan dijalankan pada sumber untuk mengekstrak data. Misalnya, select * from sampledb.sample_table. Diperlukan jika tidak menggunakan partisi.
table	Tabel yang datanya akan diekstrak. Parameter ini juga menerima subkueri di dalam tanda kurung. Misalnya Person atau (select id, name from Person) as subq. Diperlukan saat menggunakan partisi.
partitionColumn	Nama kolom yang akan digunakan untuk partisi. Hanya kolom numerik yang didukung. Diperlukan saat menggunakan partisi.
connectionProperties	Opsional: String properti yang akan digunakan untuk koneksi JDBC. Format string harus [propertyName=property;]*. Contoh, connectTimeout=60;socketTimeout=600. Untuk mengetahui informasi selengkapnya, lihat Parameter Koneksi dalam dokumentasi PostgreSQL.
username	Opsional: Nama pengguna yang akan digunakan untuk koneksi JDBC. Anda dapat meneruskan nilai ini yang dienkripsi oleh kunci Cloud KMS sebagai string berenkode Base64.
password	Opsional: Sandi yang akan digunakan untuk koneksi JDBC. Anda dapat meneruskan nilai ini yang dienkripsi oleh kunci Cloud KMS sebagai string berenkode Base64.
KMSEncryptionKey	Opsional: Kunci enkripsi Cloud KMS yang akan digunakan untuk mendekripsi nama pengguna, sandi, dan string koneksi. Jika meneruskan kunci Cloud KMS, Anda juga harus mengenkripsi nama pengguna, sandi, dan string koneksi.
numPartitions	Opsional: Jumlah partisi yang akan digunakan. Jika tidak ditentukan, angka konservatif akan diasumsikan oleh pekerja.
disabledAlgorithms	Opsional: Algoritma yang dipisahkan koma yang akan dinonaktifkan. Jika nilai ini disetel ke none, tidak ada algoritma yang dinonaktifkan. Gunakan parameter ini dengan hati-hati, karena algoritma yang dinonaktifkan secara default mungkin memiliki masalah kerentanan atau performa. Contoh: SSLv3, RC4.
extraFilesToStage	Jalur Cloud Storage yang dipisahkan koma atau rahasia Secret Manager untuk file yang akan di-stage di worker. File ini disimpan di direktori /extra_files di setiap pekerja. Contoh, gs://<my-bucket>/file.txt,projects/<project-id>/secrets/<secret-id>/versions/<version-id>.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Job name, masukkan nama pekerjaan yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih the PostgreSQL to BigQuery template.
5. Di kolom parameter yang disediakan, masukkan parameter value Anda.
6. Klik Run job.

gcloud

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/PostgreSQL_to_BigQuery \
    --parameters \
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

Ganti kode berikut:

• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• JDBC_CONNECTION_URL: URL koneksi JDBC
• SOURCE_SQL_QUERY: kueri SQL yang akan dijalankan pada database sumber
• DATASET: set data BigQuery Anda
• TABLE_NAME: nama tabel BigQuery Anda
• PATH_TO_TEMP_DIR_ON_GCS: jalur Cloud Storage Anda ke direktori sementara
• CONNECTION_PROPERTIES: properti koneksi JDBC, jika diperlukan
• CONNECTION_USERNAME: nama pengguna koneksi JDBC
• CONNECTION_PASSWORD: sandi koneksi JDBC
• KMS_ENCRYPTION_KEY: kunci enkripsi Cloud KMS

API

Untuk menjalankan template menggunakan REST API, kirim permintaan HTTP POST. Untuk informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launchParameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/PostgreSQL_to_BigQuery"
    "parameters": {
      "connectionURL": "JDBC_CONNECTION_URL",
      "query": "SOURCE_SQL_QUERY",
      "outputTable": "PROJECT_ID:DATASET.TABLE_NAME",
      "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
      "connectionProperties": "CONNECTION_PROPERTIES",
      "username": "CONNECTION_USERNAME",
      "password": "CONNECTION_PASSWORD",
      "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
    },
    "environment": { "zone": "us-central1-f" }
  }
}

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• JDBC_CONNECTION_URL: URL koneksi JDBC
• SOURCE_SQL_QUERY: kueri SQL yang akan dijalankan pada database sumber
• DATASET: set data BigQuery Anda
• TABLE_NAME: nama tabel BigQuery Anda
• PATH_TO_TEMP_DIR_ON_GCS: jalur Cloud Storage Anda ke direktori sementara
• CONNECTION_PROPERTIES: properti koneksi JDBC, jika diperlukan
• CONNECTION_USERNAME: nama pengguna koneksi JDBC
• CONNECTION_PASSWORD: sandi koneksi JDBC
• KMS_ENCRYPTION_KEY: kunci enkripsi Cloud KMS

Kode sumber template

Java

/*
 * Copyright (C) 2018 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.templates;

import static com.google.cloud.teleport.v2.utils.GCSUtils.getGcsFileAsString;
import static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;

import com.google.api.services.bigquery.model.TableRow;
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.options.JdbcToBigQueryOptions;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.utils.JdbcConverters;
import com.google.common.annotations.VisibleForTesting;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.values.PCollection;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/jdbc-to-googlecloud/README_Jdbc_to_BigQuery_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Jdbc_to_BigQuery_Flex",
    category = TemplateCategory.BATCH,
    displayName = "JDBC to BigQuery with BigQuery Storage API support",
    description = {
      "The JDBC to BigQuery template is a batch pipeline that copies data from a relational database table into an existing BigQuery table. "
          + "This pipeline uses JDBC to connect to the relational database. You can use this template to copy data from any relational database with available JDBC drivers into BigQuery.",
      "For an extra layer of protection, you can also pass in a Cloud KMS key along with a Base64-encoded username, password, and connection string parameters encrypted with the Cloud KMS key. "
          + "See the <a href=\"https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt\">Cloud KMS API encryption endpoint</a> for additional details on encrypting your username, password, and connection string parameters."
    },
    optionsClass = JdbcToBigQueryOptions.class,
    flexContainerName = "jdbc-to-bigquery",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/jdbc-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "The BigQuery table must exist before pipeline execution.",
      "The BigQuery table must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class JdbcToBigQuery {

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link JdbcToBigQuery#run} method to start the pipeline
   * and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    // Parse the user options passed from the command-line
    JdbcToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(JdbcToBigQueryOptions.class);

    run(options, writeToBQTransform(options));
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @param writeToBQ the transform that outputs {@link TableRow}s to BigQuery.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(JdbcToBigQueryOptions options, Write<TableRow> writeToBQ) {
    // Validate BQ STORAGE_WRITE_API options
    BigQueryIOUtils.validateBQStorageApiOptionsBatch(options);

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(options.getConnectionURL(), options.getKMSEncryptionKey()))
            .withUsername(maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()))
            .withPassword(maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()));

    if (options.getDriverJars() != null) {
      dataSourceConfiguration = dataSourceConfiguration.withDriverJars(options.getDriverJars());
    }

    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    /*
     * Step 1: Read records via JDBC and convert to TableRow
     *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
     */
    PCollection<TableRow> rows;
    if (options.getPartitionColumn() != null && options.getTable() != null) {
      // Read with Partitions
      // TODO(pranavbhandari): Support readWithPartitions for other data types.
      JdbcIO.ReadWithPartitions<TableRow, Long> readIO =
          JdbcIO.<TableRow>readWithPartitions()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withTable(options.getTable())
              .withPartitionColumn(options.getPartitionColumn())
              .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
      if (options.getNumPartitions() != null) {
        readIO = readIO.withNumPartitions(options.getNumPartitions());
      }
      if (options.getLowerBound() != null && options.getUpperBound() != null) {
        readIO =
            readIO.withLowerBound(options.getLowerBound()).withUpperBound(options.getUpperBound());
      }

      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JDBC with Partitions", readIO);
    } else {
      JdbcIO.Read<TableRow> readIO =
          JdbcIO.<TableRow>read()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withQuery(options.getQuery())
              .withCoder(TableRowJsonCoder.of())
              .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));

      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JdbcIO", readIO);
    }

    /*
     * Step 2: Append TableRow to an existing BigQuery table
     */
    rows.apply("Write to BigQuery", writeToBQ);

    // Execute the pipeline and return the result.
    return pipeline.run();
  }

  /**
   * Create the {@link Write} transform that outputs the collection to BigQuery as per input option.
   */
  @VisibleForTesting
  static Write<TableRow> writeToBQTransform(JdbcToBigQueryOptions options) {
    Write<TableRow> write =
        BigQueryIO.writeTableRows()
            .withoutValidation()
            .withCreateDisposition(Write.CreateDisposition.valueOf(options.getCreateDisposition()))
            .withWriteDisposition(
                options.getIsTruncate()
                    ? Write.WriteDisposition.WRITE_TRUNCATE
                    : Write.WriteDisposition.WRITE_APPEND)
            .withCustomGcsTempLocation(
                StaticValueProvider.of(options.getBigQueryLoadingTemporaryDirectory()))
            .to(options.getOutputTable());

    if (Write.CreateDisposition.valueOf(options.getCreateDisposition())
        != Write.CreateDisposition.CREATE_NEVER) {
      write = write.withJsonSchema(getGcsFileAsString(options.getBigQuerySchemaPath()));
    }

    return write;
  }
}

Langkah selanjutnya

• Pelajari Template Dataflow.
• Lihat daftar template yang disediakan Google.