Sardineでは、あらゆる場面でprotocol buffers(protobuf)を使用しています。これは、軽量で言語に依存しない、型付きのシリアライゼーション形式です。
protobufを使ってBigQueryテーブルのデータスキーマを定義し、同じprotobufファイルからバックエンドサーバー(go)のコードも生成しています。バックエンドサーバーは(protobufから生成された)golangのstructにデータを入れてpubsubに送信し、その後dataflowジョブがそれをBigQueryに取り込みます。
このシンプルなパターンはうまく機能しましたが、一つ注意点がありました。開始当初、pubsubオブジェクトをBigQueryの行挿入に変換する明確な方法がなかったため、あるオブジェクトから別のオブジェクトへデータをコピーするために、以下のような定型コードを書く必要がありました。
TableRow clientMetadataRow = new TableRow() .set("session_key", clientMetadata.getSessionKey()) .set("client_id", clientMetadata.getClientId()) .set("revision", clientMetadata.getRevision()) .set("user_id", clientMetadata.getUserId())...
これは非常に手間がかかり、ミスも起きやすいものでした。BigQueryテーブルに新しいフィールドを追加するたびに、上記の定型コードも更新する必要がありました。もっと良い方法があるはずでした。
(補足: もし今日新しいサービスを立ち上げるなら、pubsubのシリアライゼーション形式としてprotobufの代わりにJSONを使い、Google公式のpubsub-to-bigqueryジョブテンプレート(ベータ版)を使うという選択肢もあります —https://cloud.google.com/dataflow/docs/guides/templates/provided-streaming)
実はprotobufオブジェクトは動的にイテレートできることが分かったため、フィールド名をハードコーディングする代わりに、以下のようにすることができました。
void copyFields(GeneratedMessageV3 fromProto, TableRow toRow) { allFields = fromProto.getDescriptorForType().getFields(); for (Descriptors.FieldDescriptor field: allFields) { Object value = fields.get(field); String columnName = field.getName(); switch (field.getJavaType()) { case STRING: if (value != null) { toRow.set(columnName, value); } break;...
これだけです!これにより定型コードが削減され、Javaコードを更新する心配がなくなりました(なお、スキーマを変更した場合は、ジョブランナーが最新のproto定義を認識できるように、dataflowジョブを再デプロイする必要があります)
コミュニティが車輪の再発明をしなくて済むように、以下のリポジトリでコードを公開しました — https://github.com/sardine-ai/proto-to-bq-java
ご意見・ご感想がありましたら、こちらまでお気軽にご連絡ください https://twitter.com/kazukinishiura
現在、あらゆるエンジニアリング職種で採用を行っています — https://www.sardine.ai/careers