Use Apache Beam and SpannerIO to connect to Spanner Omni

Apache Beam is an open source, unified model for defining both batch and streaming data-parallel processing pipelines. This document describes how to use the SpannerIO connector within an Apache Beam pipeline to read from or write to Spanner Omni databases.

Before you begin

To connect SpannerIO to Spanner Omni, ensure that you meet the following requirements:

  • Initialize a database within your Spanner Omni environment.

  • If you use encryption, make sure you use a compatible version of Apache Beam:

    • For TLS encryption, version 2.69.0 or later.
    • For mutual TLS (mTLS) encryption, version 2.75.0 or later.

  • Set up authentication credentials for your environment.

Configure SpannerIO to connect to Spanner Omni

To connect SpannerIO to Spanner Omni, configure SpannerConfig with your database details and connection parameters.

To configure the connection, choose one of the following connection modes:

Connect using plain-text communication

To establish a plain-text connection, specify the Spanner Omni endpoint, enable experimental host support with the withExperimentalHost() method, and configure the pipeline with the withUsingPlainTextChannel() method.

The following example shows how to configure a plain-text connection:

SpannerConfig spannerConfig =
    SpannerConfig.create()
        .withDatabaseId("DATABASE_ID")
        // Define the Spanner Omni endpoint
        .withExperimentalHost("http://ENDPOINT")
        // Use a plain-text connection
        .withUsingPlainTextChannel(true);

Replace the following:

  • DATABASE_ID: the ID of your Spanner Omni database, for example, test-db.

  • ENDPOINT: the endpoint of your Spanner Omni instance, for example, localhost:15000.

Connect using encryption

To protect database traffic and ensure secure communication between Apache Beam and Spanner Omni, you can connect using TLS or mTLS encryption. Using encryption helps keep your credentials and data confidential.

Use TLS encryption

To protect database traffic between Apache Beam and Spanner Omni using TLS encryption, you don't need to specify credential properties in SpannerConfig. Instead, configure a Java truststore with the Spanner Omni CA certificate, and then configure the SpannerConfig to use the secure TLS endpoint.

Step 1: Configure the Java truststore

To secure communication, you must import the CA certificate generated by Spanner Omni into a Java truststore. Use one of the following options:

Default Java truststore

Add the CA certificate generated by Spanner Omni to the standard Java truststore by running the following command:

sudo keytool -import -trustcacerts \
  -file ~/.spanner/certs/ca.crt \
  -alias spanner-ca \
  -keystore $JAVA_HOME/lib/security/cacerts

Custom truststore

To ensure that your pipeline can still connect to other databases or services that use standard certificate authorities (CAs), create a custom truststore:

  1. Create a custom truststore by copying the existing Java truststore:

    cp $JAVA_HOME/lib/security/cacerts PATH_TO_CUSTOM_CA_CERTIFICATE

  2. Import the CA certificate into your custom truststore:

    keytool -import -trustcacerts \
  -file ~/.spanner/certs/ca.crt \
  -alias spanner-ca \
  -keystore PATH_TO_CUSTOM_CA_CERTIFICATE

  3. Pass the custom CA certificate store when you run the pipeline:

    java -Djavax.net.ssl.trustStore=PATH_TO_CUSTOM_CA_CERTIFICATE \
  -Djavax.net.ssl.trustStorePassword=changeit \
  -jar PIPELINE_NAME.jar

Replace the following:

  • PATH_TO_CUSTOM_CA_CERTIFICATE: the path to your custom CA certificate store.

  • PIPELINE_NAME: the name of your Apache Beam pipeline.

Step 2: Configure SpannerConfig

To configure SpannerConfig to use a secure TLS connection, add the following code to your pipeline:

SpannerConfig spannerConfig =
    SpannerConfig.create()
        .withDatabaseId("DATABASE_ID")
        // Define the secure Spanner Omni endpoint
        .withExperimentalHost("https://ENDPOINT");

Replace the following:

  • DATABASE_ID: the ID of your Spanner Omni database, for example, test-db.

  • ENDPOINT: the endpoint of your Spanner Omni instance, for example, localhost:15000.

Use mTLS encryption

To establish a mutual TLS (mTLS) connection using Apache Beam, you must configure the Java truststore with the CA certificate, generate or convert a client private key to PKCS#8 format, and then configure SpannerConfig with the client certificate and key path.

Step 1: Configure the Java truststore

Configure the Java truststore with the Spanner Omni CA certificate as described in Step 1: Configure the Java truststore earlier in this document.

Step 2: Convert or generate the client private key

To connect using mTLS, ensure that your client private key is in the PKCS#8 format. Use one of the following options:

openssl

To convert the client key generated by Spanner Omni to a format compliant with Java, run the following command:

openssl pkcs8 -topk8 \
  -in ~/.spanner/certs/client.key \
  -out ~/.spanner/certs/java-client.key \
  -nocrypt

Spanner Omni CLI

Generate the key directly in PKCS#8 format when creating your client certificate by using the Spanner Omni CLI with the --generate-pkcs8-key flag.

To generate a client certificate and client private key in PKCS#8 format, run the following command:

spanner certificates create-client CLIENT_NAME \
  --ca-certificate-directory=PATH_TO_CA_CERTIFICATES \
  --ca-private-key-directory=PATH_TO_PRIVATE_KEYS \
  --output-directory=PATH_TO_CERTIFICATES \
  --generate-pkcs8-key

Replace the following:

  • CLIENT_NAME: the name of the client to generate the certificate and private key for.

  • PATH_TO_CA_CERTIFICATES: the path to the directory containing the CA certificates.

  • PATH_TO_PRIVATE_KEYS: the path to the directory containing the CA private keys.

  • PATH_TO_CERTIFICATES: the path to the directory where the client certificate and private key are saved.

Step 3: Configure SpannerConfig

Configure the SpannerConfig in your pipeline code with the client certificate and the client private key:

SpannerConfig spannerConfig =
    SpannerConfig.create()
        .withDatabaseId("DATABASE_ID")
        // Define the secure Spanner Omni endpoint
        .withExperimentalHost("https://ENDPOINT")
        // Specify the paths to the client certificate and private key
        .withClientCert(
            "PATH_TO_CLIENT_CERT",
            "PATH_TO_CLIENT_CERT_KEY");

Replace the following:

  • DATABASE_ID: the ID of your Spanner Omni database, for example, test-db.

  • ENDPOINT: the endpoint of your Spanner Omni instance, for example, localhost:15000.

  • PATH_TO_CLIENT_CERT: the path to your client certificate file.

  • PATH_TO_CLIENT_CERT_KEY: the path to your client private key file.

Configure authentication tokens

Authentication tokens aren't recommended for client use because the tokens generated by Spanner Omni expire and require manual renewal with the Spanner Omni CLI. To use an authentication token with a TLS or mTLS setup for a Spanner Omni endpoint, set the SPANNER_EXPERIMENTAL_HOST_AUTH_TOKEN environment variable to the value of the authentication token generated by the Spanner Omni CLI. Leave this variable unset for connections that don't require credentials.

What's next