DBeam

A connector tool to extract data from SQL databases and import into GCS using Apache Beam.

This tool is runnable locally, or on any other backend supported by Apache Beam, e.g. Cloud Dataflow.

DEVELOPMENT STATUS: Beta, with production usage since August 2017.

Overview

DBeam is tool based that reads all the data from single SQL database table, converts the data into Avro and stores it into appointed location, usually in GCS. It runs as a single threaded Apache Beam pipeline.

DBeam requires the database credentials, the database table name to read, and the output location to store the extracted data into. DBeam first makes a single select into the target table with limit one to infer the table schema. After the schema is created the job will be launched which simply streams the table contents via JDBC into target location as Avro.

dbeam-core package features

• Supports both PostgreSQL and MySQL JDBC connectors
• Supports Google CloudSQL managed databases
• Currently output only to Avro format
• Reads database from an external password file (

  --passwordFile

  ) or an external KMS encrypted password file (

  --passwordFileKmsEncrypted

  )
• Can filter only records of the current day with the

  --partitionColumn

  parameter
• Check and fail on too old partition dates. Snapshot dumps are not filtered by a given date/partition, when running for a too old partition, the job fails to avoid new data in old partitions. (can be disabled with

  --skipPartitionCheck

  )
• Implemented as Apache Beam SDK pipeline, supporting any of its runners (tested with

  DirectRunner

  and

  DataflowRunner

  )

DBeam export parameters

com.spotify.dbeam.options.DBeamPipelineOptions:

  --connectionUrl=
    The JDBC connection url to perform the export.
  --password=
    Plaintext password used by JDBC connection.
  --passwordFile=
    A path to a file containing the database password.
  --passwordFileKmsEncrypted=
    A path to a file containing the database password, KMS encrypted and base64
    encoded.
  --sqlFile=
    A path to a file containing a SQL query (used instead of --table parameter).
  --table=
    The database table to query and perform the export.
  --username=
    Default: dbeam-extractor
    The database user name used by JDBC to authenticate.
com.spotify.dbeam.options.OutputOptions:
  --output=
    The path for storing the output.
  --dataOnly=
    Default: false
    Store only the data files in output folder, skip queries, metrics and
    metadata files.
com.spotify.dbeam.options.JdbcExportPipelineOptions:
    Configures the DBeam SQL export
  --avroCodec=
    Default: deflate6
    Avro codec (e.g. deflate6, deflate9, snappy).
  --avroDoc=
    The top-level record doc string of the generated avro schema.
  --avroSchemaFilePath=
    Path to file with a target AVRO schema.
  --avroSchemaName=
    The name of the generated avro schema. By default it uses the table name.
  --avroSchemaNamespace=
    Default: dbeam_generated
    The namespace of the generated avro schema.
  --exportTimeout=
    Default: P7D
    Export timeout, after this duration the job is cancelled and the export
    terminated.
  --fetchSize=
    Default: 10000
    Configures JDBC Statement fetch size.
  --limit=
    Limit the output number of rows, indefinite by default.
  --minPartitionPeriod=
    The minimum partition required for the job not to fail (when partition
    column is not specified),by default now() - 2*partitionPeriod.
  --partition=
    The date/timestamp of the current partition.
  --partitionColumn=
    The name of a date/timestamp column to filter data based on current
    partition.
  --partitionPeriod=
    The period frequency which the export runs, used to filter based on current
    partition and also to check if exports are running for too old partitions.
  --preCommand=
    SQL commands to be executed before query.
  --queryParallelism=
    Max number of queries to run in parallel for exports. Single query used if
    nothing specified. Should be used with splitColumn.
  --skipPartitionCheck=
    Default: false
    When partition column is not specified, fails if partition is too old; set
    this flag to ignore this check.
  --splitColumn=
    A long/integer column used to create splits for parallel queries. Should be
    used with queryParallelism.
  --useAvroLogicalTypes=
    Default: false
    Controls whether generated Avro schema will contain logicalTypes or not.

Input Avro schema file

If provided an input Avro schema file, dbeam will read input schema file and use some of the properties when an output Avro schema is created.

Following fields will be propagated from input into output schema:

• record.doc

• record.namespace

• record.field.doc

DBeam Parallel Mode

This is a pre-alpha feature currently under development and experimentation.

Read queries used by dbeam to extract data generally don't place any locks, and hence multiple read queries can run in parallel. When running in parallel mode with

--queryParallelism

--splitColumn

queryParallelism

splitColumn

Recommended usage: Beam would run each query generated by DBeam in 1 dedicated vCPU (when running with Dataflow Runner), thus for best performance it is recommended that the total number of vCPU available for a given job should be equal to the

queryParallelism

workerMachineType

n1-standard-w

numWorkers

n

queryParallelism

q

n*w

q = n * w

For an export of a table running from a dedicated PostgresQL replica, we have seen best performance over vCPU time and wall time when having a

queryParallelism

queryParallelism

queryParallelism

Building

Building and testing can be achieved with

mvn

mvn verify

In order to create a jar with all dependencies under

./dbeam-core/target/dbeam-core-shaded.jar

mvn clean package -Ppack

Usage examples

Using Java from the command line:

java -cp ./dbeam-core/target/dbeam-core-shaded.jar \
  com.spotify.dbeam.jobs.JdbcAvroJob \
  --output=gs://my-testing-bucket-name/ \
  --username=my_database_username \
  --password=secret \
  --connectionUrl=jdbc:postgresql://some.database.uri.example.org:5432/my_database \
  --table=my_table

For CloudSQL:

java -cp ./dbeam-core/target/dbeam-core-shaded.jar \
  com.spotify.dbeam.jobs.JdbcAvroJob \
  --output=gs://my-testing-bucket-name/ \
  --username=my_database_username \
  --password=secret \
  --connectionUrl=jdbc:postgresql://google/database?socketFactory=com.google.cloud.sql.postgres.SocketFactory&socketFactoryArg=project:region:cloudsql-instance \
  --table=my_table

• When using MySQL:

  --connectionUrl=jdbc:mysql://google/database?socketFactory=com.google.cloud.sql.mysql.SocketFactory&cloudSqlInstance=project:region:cloudsql-instance&useCursorFetch=true

• Note

  ?useCursorFetch=true

  is important for MySQL, to avoid early fetching all rows, more details on MySQL docs.
• More details can be found at CloudSQL JDBC SocketFactory

To run a cheap data extraction, as a way to validate, one can add

--limit=10 --skipPartitionCheck

Password configuration

Database password can be configured by simply passing

--password=writepasswordhere

--passwordFile=/path/to/file/containing/password

--passwordFile=gs://gcs-bucket/path/to/file/containing/password

A more robust configuration is to point to a Google KMS encrypted file. DBeam will try to decrypt using KMS if the file ends with

.encrypted

--passwordFileKmsEncrypted=gs://gcs-bucket/path/to/db-password.encrypted

The file should contain a base64 encoded encrypted content. It can be generated using

gcloud

echo -n "super_secret_password" \
  | gcloud kms encrypt \
      --location "global" \
      --keyring "dbeam" \
      --key "default" \
      --project "mygcpproject" \
      --plaintext-file - \
      --ciphertext-file - \
  | base64 \
  | gsutil cp - gs://gcs-bucket/path/to/db-password.encrypted

KMS location, keyring, and key can be configured via Java Properties, defaults are:

java \
  -DKMS_KEYRING=dbeam \
  -DKMS_KEY=default \
  -DKMS_LOCATION=global \
  -DKMS_PROJECT=default_gcp_project \
  -cp ./dbeam-core/target/dbeam-core-shaded.jar \
  com.spotify.dbeam.jobs.JdbcAvroJob \
  ...

Using as a library

To include DBeam library in a mvn project add the following dependency in

pom.xml

  com.spotify
  dbeam-core
  ${dbeam.version}

To include DBeam library in a SBT project add the following dependency in

build.sbt

  libraryDependencies ++= Seq(
   "com.spotify" % "dbeam-core" % dbeamVersion
  )

Development

Make sure you have mvn installed. For editor, IntelliJ IDEA is recommended.

To test and verify changes during development, run:

mvn verify

Or:

mvn verify -Pcoverage

This project adheres to the Open Code of Conduct. By participating, you are expected to honor this code.

Release

Every push to master will deploy a snapshot version to Sonatype. You can check the deployment in the following links:

• https://github.com/spotify/dbeam/actions
• https://oss.sonatype.org/#nexus-search;quick~dbeam-core

License

Copyright 2016-2020 Spotify AB.

Licensed under the Apache License, Version 2.0: http://www.apache.org/licenses/LICENSE-2.0