Java Database Connectivity (JDBC) ke template BigQuery

Template JDBC ke BigQuery adalah pipeline batch yang menyalin data dari tabel database relasional ke dalam tabel BigQuery yang sudah ada. Pipeline ini menggunakan JDBC untuk terhubung ke database relasional. Gunakan template ini untuk menyalin data dari database relasional apa pun dengan driver JDBC yang tersedia ke BigQuery.

Sebagai lapisan perlindungan tambahan, Anda dapat meneruskan kunci Cloud KMS, beserta nama pengguna, sandi, dan parameter string koneksi yang dienkode dengan Base64 yang dienkripsi dengan kunci Cloud KMS. Untuk mengetahui detail tambahan tentang enkripsi nama pengguna, sandi, dan parameter string koneksi Anda, lihat endpoint enkripsi API Cloud KMS.

Persyaratan pipeline

Driver JDBC untuk database relasional harus tersedia.
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
`driverJars`	Daftar file JAR driver yang dipisahkan koma. Contoh: `gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar`.
`driverClassName`	Nama class driver JDBC. Misalnya: `com.mysql.jdbc.Driver`.
`connectionURL`	String URL koneksi JDBC. Misalnya, `jdbc:mysql://some-host:3306/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. Perhatikan perbedaan antara string koneksi database non-RAC Oracle (`jdbc:oracle:thin:@some-host:<port>:<sid>`) dan string koneksi database RAC Oracle (`jdbc:oracle:thin:@//some-host[:<port>]/<service_name>`). Misalnya: `jdbc:mysql://some-host:3306/sampledb`.
`outputTable`	Lokasi tabel BigQuery yang akan menjadi tujuan penulisan output. Nama harus menggunakan format `<project>:<dataset>.<table_name>`. Skema tabel harus cocok dengan objek input. Contoh: `<my-project>:<my-dataset>.<my-table>`.
`bigQueryLoadingTemporaryDirectory`	Direktori sementara untuk proses pemuatan BigQuery. Contoh: `gs://your-bucket/your-files/temp_dir`.
`connectionProperties`	Opsional: String properti yang akan digunakan untuk koneksi JDBC. Gunakan format string [propertyName=property;]*. Contoh: `unicode=true;characterEncoding=UTF-8`.
`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.
`query`	Opsional: Kueri yang akan dijalankan di sumber untuk mengekstrak data. Misalnya: `select * from sampledb.sample_table`.
`KMSEncryptionKey`	Opsional: Kunci enkripsi Cloud KMS yang akan digunakan untuk mendekripsi nama pengguna, sandi, dan string koneksi. Jika Anda meneruskan kunci Cloud KMS, nama pengguna, sandi, dan string koneksi harus diteruskan secara terenkripsi. Contoh: `projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key`.
`useColumnAlias`	Opsional: Jika diaktifkan (ditetapkan ke `true`), pipeline akan menggunakan alias kolom ("AS"), bukan nama kolom, untuk memetakan baris ke BigQuery. Nilai defaultnya adalah `false`.
`isTruncate`	Opsional: Jika diaktifkan (ditetapkan ke `true`), pipeline akan terpotong sebelum memuat data ke BigQuery. Defaultnya adalah `false`, yang menyebabkan pipeline menambahkan data.
`partitionColumn`	Opsional: Jika parameter ini disediakan (bersama dengan `table`), JdbcIO akan membaca tabel secara paralel dengan menjalankan beberapa instance kueri di tabel (subkueri) yang sama menggunakan rentang. Saat ini, hanya mendukung kolom partisi `Long`.
`table`	Opsional: Tabel untuk dibaca saat menggunakan partisi. Parameter ini juga menerima subkueri di dalam tanda kurung. Misalnya: `(select id, name from Person` sebagai subq).
`numPartitions`	Opsional: Jumlah partisi. Dengan batas bawah dan atas, nilai ini membentuk langkah partisi untuk ekspresi klausa `WHERE` yang dihasilkan yang digunakan untuk membagi kolom partisi secara merata. Jika input kurang dari `1`, angka ditetapkan ke `1`.
`lowerBound`	Opsional: Batas bawah untuk digunakan dalam skema partisi. Jika tidak diberikan, nilai ini akan otomatis disimpulkan oleh Apache Beam untuk jenis yang didukung.
`upperBound`	Opsional: Batas atas untuk digunakan dalam skema partisi. Jika tidak diberikan, nilai ini akan otomatis disimpulkan oleh Apache Beam untuk jenis yang didukung.
`disabledAlgorithms`	Opsional: Algoritma yang dipisahkan koma yang akan dinonaktifkan. Jika nilai ini disetel ke `none`, tidak ada algoritma yang dinonaktifkan. Gunakan dengan hati-hati, karena algoritma yang dinonaktifkan secara default diketahui memiliki masalah kerentanan atau performa. Contoh: `SSLv3, RC4`.
`extraFilesToStage`	Opsional: 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` pada setiap pekerja. Contoh: `gs://your-bucket/file.txt,projects/project-id/secrets/secret-id/versions/version-id`.
`useStorageWriteApi`	Opsional: Jika `true`, pipeline akan menggunakan BigQuery Storage Write API. Nilai defaultnya adalah `false`. Untuk informasi selengkapnya, lihat Menggunakan Storage Write API.
`useStorageWriteApiAtLeastOnce`	Opsional: Saat menggunakan Storage Write API, menentukan semantik penulisan. Untuk menggunakan semantik minimal satu kali , tetapkan parameter ini ke `true`. Untuk menggunakan semantik tepat satu kali, tetapkan parameter ke `false`. Parameter ini hanya berlaku jika `useStorageWriteApi` adalah `true`. Nilai defaultnya adalah `false`.

Menjalankan template

Konsol

Buka halaman Create job from template Dataflow.

Buka Buat tugas dari template

Di kolom Job name, masukkan nama pekerjaan yang unik.
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.
Dari menu drop-down Dataflow template, pilih the JDBC to BigQuery template.
Di kolom parameter yang disediakan, masukkan parameter value Anda.
Klik Run job.

gcloud

Di shell atau terminal Anda, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Jdbc_to_BigQuery_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       driverJars=DRIVER_JARS,\
       driverClassName=DRIVER_CLASS_NAME,\
       connectionURL=CONNECTION_URL,\
       outputTable=OUTPUT_TABLE,\
       bigQueryLoadingTemporaryDirectory=BIG_QUERY_LOADING_TEMPORARY_DIRECTORY,\

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
DRIVER_JARS: jalur Cloud Storage yang dipisahkan koma dari driver JDBC
DRIVER_CLASS_NAME: nama class driver JDBC
CONNECTION_URL: string URL koneksi JDBC.
OUTPUT_TABLE: tabel output BigQuery
BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: Direktori sementara untuk proses pemuatan BigQuery

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",
     "parameters": {
       "driverJars": "DRIVER_JARS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "CONNECTION_URL",
       "outputTable": "OUTPUT_TABLE",
       "bigQueryLoadingTemporaryDirectory": "BIG_QUERY_LOADING_TEMPORARY_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Jdbc_to_BigQuery_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

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
DRIVER_JARS: jalur Cloud Storage yang dipisahkan koma dari driver JDBC
DRIVER_CLASS_NAME: nama class driver JDBC
CONNECTION_URL: string URL koneksi JDBC.
OUTPUT_TABLE: tabel output BigQuery
BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: Direktori sementara untuk proses pemuatan BigQuery

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.