Apache Beam は、バッチとストリーミングの両方のデータの並列処理パイプラインを定義するオープンソースの統合モデルです。このドキュメントでは、Apache Beam パイプライン内で SpannerIO コネクタを使用して、Spanner Omni データベースとの間で読み取りまたは書き込みを行う方法について説明します。
始める前に
SpannerIO を Spanner Omni に接続するには、次の要件を満たしていることを確認してください。
Spanner Omni 環境内でデータベースを初期化します。
暗号化を使用する場合は、互換性のあるバージョンの Apache Beam を使用してください。
- TLS 暗号化の場合、バージョン 2.69.0 以降。
- 相互 TLS（mTLS）暗号化の場合、バージョン 2.75.0 以降。
環境の認証情報を設定します。
Spanner Omni に接続するように SpannerIO を構成する
SpannerIO を Spanner Omni に接続するには、データベースの詳細と接続パラメータを使用して SpannerConfig を構成します。
接続を構成するには、次のいずれかの接続モードを選択します。
平文通信を使用して接続する
プレーン テキスト接続を確立するには、Spanner Omni エンドポイントを指定し、withExperimentalHost() メソッドで試験運用版のホスト サポートを有効にして、withUsingPlainTextChannel() メソッドでパイプラインを構成します。
次の例は、プレーン テキスト接続を構成する方法を示しています。
SpannerConfig spannerConfig =
SpannerConfig.create()
.withDatabaseId("DATABASE_ID")
// Define the Spanner Omni endpoint
.withExperimentalHost("http://ENDPOINT")
// Use a plain-text connection
.withUsingPlainTextChannel(true);
次のように置き換えます。
DATABASE_ID: Spanner Omni データベースの ID（例:test-db）。ENDPOINT: Spanner Omni インスタンスのエンドポイント（例:localhost:15000）。
暗号化を使用して接続する
データベース トラフィックを保護し、Apache Beam と Spanner Omni 間の安全な通信を確保するには、TLS または mTLS 暗号化を使用して接続します。暗号化を使用すると、認証情報とデータの機密性を維持できます。
TLS 暗号化を使用する
TLS 暗号化を使用して Apache Beam と Spanner Omni 間のデータベース トラフィックを保護する場合、SpannerConfig で認証情報プロパティを指定する必要はありません。代わりに、Spanner Omni CA 証明書を使用して Java トラストストアを構成し、安全な TLS エンドポイントを使用するように SpannerConfig を構成します。
ステップ 1: Java トラストストアを構成する
通信を保護するには、Spanner Omni によって生成された CA 証明書を Java トラストストアにインポートする必要があります。次のいずれかのオプションを使用します。
デフォルトの Java トラストストア
次のコマンドを実行して、Spanner Omni によって生成された CA 証明書を標準の Java トラストストアに追加します。
sudo keytool -import -trustcacerts \
-file ~/.spanner/certs/ca.crt \
-alias spanner-ca \
-keystore $JAVA_HOME/lib/security/cacerts
カスタム トラストストア
パイプラインが標準の認証局（CA）を使用する他のデータベースやサービスに引き続き接続できるようにするには、カスタム トラストストアを作成します。
既存の Java トラストストアをコピーして、カスタム トラストストアを作成します。
cp $JAVA_HOME/lib/security/cacerts PATH_TO_CUSTOM_CA_CERTIFICATECA 証明書をカスタム トラストストアにインポートします。
keytool -import -trustcacerts \ -file ~/.spanner/certs/ca.crt \ -alias spanner-ca \ -keystore PATH_TO_CUSTOM_CA_CERTIFICATEパイプラインの実行時にカスタム CA 証明書ストアを渡します。
java -Djavax.net.ssl.trustStore=PATH_TO_CUSTOM_CA_CERTIFICATE \ -Djavax.net.ssl.trustStorePassword=changeit \ -jar PIPELINE_NAME.jar
次のように置き換えます。
PATH_TO_CUSTOM_CA_CERTIFICATE: カスタム CA 証明書ストアのパス。PIPELINE_NAME: Apache Beam パイプラインの名前。
ステップ ２: SpannerConfig を設定する
安全な TLS 接続を使用するように SpannerConfig を構成するには、次のコードをパイプラインに追加します。
SpannerConfig spannerConfig =
SpannerConfig.create()
.withDatabaseId("DATABASE_ID")
// Define the secure Spanner Omni endpoint
.withExperimentalHost("https://ENDPOINT");
次のように置き換えます。
DATABASE_ID: Spanner Omni データベースの ID（例:test-db）。ENDPOINT: Spanner Omni インスタンスのエンドポイント（例:localhost:15000）。
mTLS 暗号化を使用する
Apache Beam を使用して相互 TLS（mTLS）接続を確立するには、CA 証明書を使用して Java truststore を構成し、クライアント秘密鍵を生成または PKCS#8 形式に変換してから、クライアント証明書と鍵パスを使用して SpannerConfig を構成する必要があります。
ステップ 1: Java トラストストアを構成する
このドキュメントの前のステップ 1: Java トラストストアを構成するの説明に従って、Spanner Omni CA 証明書を使用して Java トラストストアを構成します。
ステップ 2: クライアントの秘密鍵を変換または生成する
mTLS を使用して接続するには、クライアントの秘密鍵が PKCS#8 形式であることを確認します。以下のいずれかの方法を選択します。
openssl
Spanner Omni で生成されたクライアント鍵を Java に準拠した形式に変換するには、次のコマンドを実行します。
openssl pkcs8 -topk8 \
-in ~/.spanner/certs/client.key \
-out ~/.spanner/certs/java-client.key \
-nocrypt
Spanner Omni CLI
--generate-pkcs8-key フラグを指定して Spanner Omni CLI を使用してクライアント証明書を作成するときに、PKCS#8 形式で鍵を直接生成します。
PKCS#8 形式のクライアント証明書とクライアント秘密鍵を生成するには、次のコマンドを実行します。
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
次のように置き換えます。
CLIENT_NAME: 証明書と秘密鍵を生成するクライアントの名前。PATH_TO_CA_CERTIFICATES: CA 証明書を含むディレクトリのパス。PATH_TO_PRIVATE_KEYS: CA 秘密鍵を含むディレクトリのパス。PATH_TO_CERTIFICATES: クライアント証明書と秘密鍵が保存されているディレクトリのパス。
ステップ 3: SpannerConfig を構成する
パイプライン コードで、クライアント証明書とクライアント秘密鍵を使用して SpannerConfig を構成します。
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");
次のように置き換えます。
DATABASE_ID: Spanner Omni データベースの ID（例:test-db）。ENDPOINT: Spanner Omni インスタンスのエンドポイント（例:localhost:15000）。PATH_TO_CLIENT_CERT: クライアント証明書ファイルのパス。PATH_TO_CLIENT_CERT_KEY: クライアントの秘密鍵ファイルのパス。
認証トークンを構成する
Spanner Omni で生成されたトークンは期限切れになり、Spanner Omni CLI で手動で更新する必要があるため、クライアントでの認証トークンの使用はおすすめしません。Spanner Omni エンドポイントの TLS または mTLS 設定で認証トークンを使用するには、SPANNER_EXPERIMENTAL_HOST_AUTH_TOKEN 環境変数を Spanner Omni CLI によって生成された認証トークンの値に設定します。認証情報を必要としない接続の場合は、この変数を設定しないでください。