Google Cloud BigQuery Storage

The BigQuery Storage API offers fast access to BigQuery-managed storage using an rpc-based protocol. It is seen as an improvement over the REST API, and bulk data extract jobs for accessing BigQuery-managed table data, but doesn’t offer any functionality around managing BigQuery resources. Further information at the official Google Cloud documentation website.

This connector communicates to the BigQuery Storage API via the gRPC protocol. The integration between Akka Stream and gRPC is handled by the Akka gRPC library. Currently, this connector only supports returning each row as an Avro GenericRecord.

Project Info: Alpakka Google Cloud BigQuery Storage
Artifact
com.lightbend.akka
akka-stream-alpakka-google-cloud-bigquery-storage
3.0.4
JDK versions
Adopt OpenJDK 8
Adopt OpenJDK 11
Scala versions2.12.11, 2.13.3
JPMS module nameakka.stream.alpakka.google.cloud.bigquery.storage
License
Readiness level
Since 3.0.4, 2021-11-12
Home pagehttps://doc.akka.io/docs/alpakka/current
Forums
Release notesGitHub releases
IssuesGithub issues
Sourceshttps://github.com/akka/alpakka

Artifacts

Akka gRPC uses Akka Discovery internally. Make sure to add Akka Discovery with the same Akka version that the application uses.

sbt
val AkkaVersion = "2.6.14"
libraryDependencies ++= Seq(
  "com.lightbend.akka" %% "akka-stream-alpakka-google-cloud-bigquery-storage" % "3.0.4",
  "com.typesafe.akka" %% "akka-stream" % AkkaVersion,
  "com.typesafe.akka" %% "akka-discovery" % AkkaVersion
)
Maven
<properties>
  <akka.version>2.6.14</akka.version>
  <scala.binary.version>2.12</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>com.lightbend.akka</groupId>
    <artifactId>akka-stream-alpakka-google-cloud-bigquery-storage_${scala.binary.version}</artifactId>
    <version>3.0.4</version>
  </dependency>
  <dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-stream_${scala.binary.version}</artifactId>
    <version>${akka.version}</version>
  </dependency>
  <dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-discovery_${scala.binary.version}</artifactId>
    <version>${akka.version}</version>
  </dependency>
</dependencies>
Gradle
def versions = [
  AkkaVersion: "2.6.14",
  ScalaBinary: "2.12"
]
dependencies {
  implementation "com.lightbend.akka:akka-stream-alpakka-google-cloud-bigquery-storage_${versions.ScalaBinary}:3.0.4"
  implementation "com.typesafe.akka:akka-stream_${versions.ScalaBinary}:${versions.AkkaVersion}"
  implementation "com.typesafe.akka:akka-discovery_${versions.ScalaBinary}:${versions.AkkaVersion}"
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Direct dependencies
OrganizationArtifactVersion
com.google.protobufprotobuf-java3.11.4
com.lightbend.akka.grpcakka-grpc-runtime_2.121.0.0
com.lightbend.akkaakka-stream-alpakka-google-common_2.123.0.4
com.thesamet.scalapbscalapb-runtime_2.120.10.4
com.typesafe.akkaakka-discovery_2.122.6.14
com.typesafe.akkaakka-http-spray-json_2.1210.1.12
com.typesafe.akkaakka-stream_2.122.6.14
io.grpcgrpc-auth1.30.0
io.grpcgrpc-stub1.30.0
org.scala-langscala-library2.12.11
Dependency tree
com.google.protobuf    protobuf-java    3.11.4
com.lightbend.akka.grpc    akka-grpc-runtime_2.12    1.0.0    Apache-2.0
    com.thesamet.scalapb    scalapb-runtime_2.12    0.10.4    Apache 2
        com.google.protobuf    protobuf-java    3.11.4
        com.lihaoyi    fastparse_2.12    2.3.0    MIT
            com.lihaoyi    geny_2.12    0.6.0    MIT
            com.lihaoyi    sourcecode_2.12    0.2.1    MIT
        com.thesamet.scalapb    lenses_2.12    0.10.4    Apache 2
            org.scala-lang.modules    scala-collection-compat_2.12    2.1.6    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang.modules    scala-collection-compat_2.12    2.1.6    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-discovery_2.12    2.6.14    Apache-2.0
        com.typesafe.akka    akka-actor_2.12    2.6.14    Apache-2.0
            com.typesafe    config    1.4.0    Apache-2.0
            org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
        com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-http2-support_2.12    10.1.12    Apache-2.0
        com.twitter    hpack    1.0.2
        com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
            com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.eclipse.jetty.alpn    alpn-api    1.1.3.v20160715
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-http_2.12    10.1.12    Apache-2.0
        com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
            com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-stream_2.12    2.6.14    Apache-2.0
        com.typesafe.akka    akka-actor_2.12    2.6.14    Apache-2.0
            com.typesafe    config    1.4.0    Apache-2.0
            org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        com.typesafe.akka    akka-protobuf-v3_2.12    2.6.14    Apache-2.0
        com.typesafe    ssl-config-core_2.12    0.4.2    Apache-2.0
            com.typesafe    config    1.4.0    Apache-2.0
            org.scala-lang.modules    scala-parser-combinators_2.12    1.1.2    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.reactivestreams    reactive-streams    1.0.3    CC0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    io.grpc    grpc-core    1.30.0    Apache 2.0
        com.google.android    annotations    4.1.1.4    Apache 2.0
        com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
        com.google.code.gson    gson    2.8.6
        com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
        com.google.guava    guava    28.2-android
            com.google.guava    failureaccess    1.0.1
            com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
            com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
            org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
        io.grpc    grpc-api    1.30.0    Apache 2.0
            com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
            com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
            com.google.guava    guava    28.2-android
                com.google.guava    failureaccess    1.0.1
                com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
                com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
                org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
            io.grpc    grpc-context    1.30.0    Apache 2.0
            org.codehaus.mojo    animal-sniffer-annotations    1.18
        io.perfmark    perfmark-api    0.19.0    Apache 2.0
            com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
        org.codehaus.mojo    animal-sniffer-annotations    1.18
    io.grpc    grpc-netty-shaded    1.30.0    Apache 2.0
        io.grpc    grpc-core    1.30.0    Apache 2.0
            com.google.android    annotations    4.1.1.4    Apache 2.0
            com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
            com.google.code.gson    gson    2.8.6
            com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
            com.google.guava    guava    28.2-android
                com.google.guava    failureaccess    1.0.1
                com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
                com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
                org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
            io.grpc    grpc-api    1.30.0    Apache 2.0
                com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
                com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
                com.google.guava    guava    28.2-android
                    com.google.guava    failureaccess    1.0.1
                    com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
                    com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
                    org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
                io.grpc    grpc-context    1.30.0    Apache 2.0
                org.codehaus.mojo    animal-sniffer-annotations    1.18
            io.perfmark    perfmark-api    0.19.0    Apache 2.0
                com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
            org.codehaus.mojo    animal-sniffer-annotations    1.18
    org.scala-lang    scala-library    2.12.11    Apache-2.0
com.lightbend.akka    akka-stream-alpakka-google-common_2.12    3.0.4
    com.github.jwt-scala    jwt-spray-json_2.12    7.1.0    Apache-2.0
        com.github.jwt-scala    jwt-json-common_2.12    7.1.0    Apache-2.0
            com.github.jwt-scala    jwt-core_2.12    7.1.0    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        io.spray    spray-json_2.12    1.3.6    Apache 2
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.google.auth    google-auth-library-credentials    0.24.1
    com.typesafe.akka    akka-http-spray-json_2.12    10.1.12    Apache-2.0
        com.typesafe.akka    akka-http_2.12    10.1.12    Apache-2.0
            com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
                com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
                    org.scala-lang    scala-library    2.12.11    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        io.spray    spray-json_2.12    1.3.6    Apache 2
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-http_2.12    10.1.12    Apache-2.0
        com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
            com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-stream_2.12    2.6.14    Apache-2.0
        com.typesafe.akka    akka-actor_2.12    2.6.14    Apache-2.0
            com.typesafe    config    1.4.0    Apache-2.0
            org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        com.typesafe.akka    akka-protobuf-v3_2.12    2.6.14    Apache-2.0
        com.typesafe    ssl-config-core_2.12    0.4.2    Apache-2.0
            com.typesafe    config    1.4.0    Apache-2.0
            org.scala-lang.modules    scala-parser-combinators_2.12    1.1.2    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.reactivestreams    reactive-streams    1.0.3    CC0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.scala-lang    scala-library    2.12.11    Apache-2.0
com.thesamet.scalapb    scalapb-runtime_2.12    0.10.4    Apache 2
    com.google.protobuf    protobuf-java    3.11.4
    com.lihaoyi    fastparse_2.12    2.3.0    MIT
        com.lihaoyi    geny_2.12    0.6.0    MIT
        com.lihaoyi    sourcecode_2.12    0.2.1    MIT
    com.thesamet.scalapb    lenses_2.12    0.10.4    Apache 2
        org.scala-lang.modules    scala-collection-compat_2.12    2.1.6    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.scala-lang.modules    scala-collection-compat_2.12    2.1.6    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.scala-lang    scala-library    2.12.11    Apache-2.0
com.typesafe.akka    akka-discovery_2.12    2.6.14    Apache-2.0
    com.typesafe.akka    akka-actor_2.12    2.6.14    Apache-2.0
        com.typesafe    config    1.4.0    Apache-2.0
        org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.scala-lang    scala-library    2.12.11    Apache-2.0
com.typesafe.akka    akka-http-spray-json_2.12    10.1.12    Apache-2.0
    com.typesafe.akka    akka-http_2.12    10.1.12    Apache-2.0
        com.typesafe.akka    akka-http-core_2.12    10.1.12    Apache-2.0
            com.typesafe.akka    akka-parsing_2.12    10.1.12    Apache-2.0
                org.scala-lang    scala-library    2.12.11    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    io.spray    spray-json_2.12    1.3.6    Apache 2
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.scala-lang    scala-library    2.12.11    Apache-2.0
com.typesafe.akka    akka-stream_2.12    2.6.14    Apache-2.0
    com.typesafe.akka    akka-actor_2.12    2.6.14    Apache-2.0
        com.typesafe    config    1.4.0    Apache-2.0
        org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    com.typesafe.akka    akka-protobuf-v3_2.12    2.6.14    Apache-2.0
    com.typesafe    ssl-config-core_2.12    0.4.2    Apache-2.0
        com.typesafe    config    1.4.0    Apache-2.0
        org.scala-lang.modules    scala-parser-combinators_2.12    1.1.2    Apache-2.0
            org.scala-lang    scala-library    2.12.11    Apache-2.0
        org.scala-lang    scala-library    2.12.11    Apache-2.0
    org.reactivestreams    reactive-streams    1.0.3    CC0
    org.scala-lang    scala-library    2.12.11    Apache-2.0
io.grpc    grpc-auth    1.30.0    Apache 2.0
    com.google.auth    google-auth-library-credentials    0.24.1
    com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
    com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
    com.google.guava    guava    28.2-android
        com.google.guava    failureaccess    1.0.1
        com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
        com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
        org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
    io.grpc    grpc-api    1.30.0    Apache 2.0
        com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
        com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
        com.google.guava    guava    28.2-android
            com.google.guava    failureaccess    1.0.1
            com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
            com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
            org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
        io.grpc    grpc-context    1.30.0    Apache 2.0
        org.codehaus.mojo    animal-sniffer-annotations    1.18
    org.codehaus.mojo    animal-sniffer-annotations    1.18
io.grpc    grpc-stub    1.30.0    Apache 2.0
    com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
    com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
    com.google.guava    guava    28.2-android
        com.google.guava    failureaccess    1.0.1
        com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
        com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
        org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
    io.grpc    grpc-api    1.30.0    Apache 2.0
        com.google.code.findbugs    jsr305    3.0.2    The Apache Software License, Version 2.0
        com.google.errorprone    error_prone_annotations    2.3.4    Apache 2.0
        com.google.guava    guava    28.2-android
            com.google.guava    failureaccess    1.0.1
            com.google.guava    listenablefuture    9999.0-empty-to-avoid-conflict-with-guava
            com.google.j2objc    j2objc-annotations    1.3    The Apache Software License, Version 2.0
            org.checkerframework    checker-compat-qual    2.5.5    GNU General Public License, version 2 (GPL2), with the classpath exception
        io.grpc    grpc-context    1.30.0    Apache 2.0
        org.codehaus.mojo    animal-sniffer-annotations    1.18
    org.codehaus.mojo    animal-sniffer-annotations    1.18
org.scala-lang    scala-library    2.12.11    Apache-2.0

Build setup

The Alpakka Google Cloud BigQuery Storage library contains the classes generated from Google’s protobuf specification.

ALPN on JDK 8

HTTP/2 requires ALPN negotiation, which comes with the JDK starting with version 8u251.

For older versions of the JDK you will need to load the jetty-alpn-agent yourself, but we recommend upgrading.

Configuration

The BigQuery Storage connector shares its basic configuration with all the Google connectors in Alpakka.

Example Test Configuration

alpakka.google.cloud.bigquery.grpc {
  host = "localhost"
  port = 21000
  rootCa = "none"
  callCredentials = "none"
}

For more configuration details consider the underlying configuration for Akka gRPC.

A manually initialized akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.GrpcBigQueryStorageReaderakka.stream.alpakka.googlecloud.bigquery.storage.javadsl.GrpcBigQueryStorageReader can be used by providing it as an attribute to the stream:

Scala
sourceval reader: GrpcBigQueryStorageReader = GrpcBigQueryStorageReader(BigQueryStorageSettings("localhost", 8000))
val sourceForReader: Source[(ReadSession.Schema, Seq[Source[ReadRowsResponse.Rows, NotUsed]]), Future[NotUsed]] =
  BigQueryStorage
    .create("projectId", "datasetId", "tableId", DataFormat.AVRO)
    .withAttributes(
      BigQueryStorageAttributes.reader(reader)
    )
Java
sourceGrpcBigQueryStorageReader reader =
    GrpcBigQueryStorageReader.apply(BigQueryStorageSettings.apply("localhost", 8000), sys);

Source<
        Tuple2<
            com.google.cloud.bigquery.storage.v1.stream.ReadSession.Schema,
            List<Source<ReadRowsResponse.Rows, NotUsed>>>,
        CompletionStage<NotUsed>>
    sourceForReader =
        BigQueryStorage.create(
                "projectId", "datasetId", "tableId", DataFormat.AVRO, readOptions, 1)
            .withAttributes(BigQueryStorageAttributes.reader(reader));

Reading

We can read in a number of ways. To read data from a table a read session needs to be created. On the session creation we can specify the number of streams to be used in order to transfer the data, this makes it feasible to achieve parallelism while ingesting the data, thus achieving better performance. To create a session the data format needs to be specified. The options provided are Avro and Arrow.

If no TableReadOptions are specified all the table’s columns shall be retrieved as a Source containing a Source for each stream, which will each deliver a section of the rows:

Scala
sourceimport akka.NotUsed
import com.google.cloud.bigquery.storage.v1.storage.ReadRowsResponse
import com.google.cloud.bigquery.storage.v1.DataFormat
import com.google.cloud.bigquery.storage.v1.stream.ReadSession
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.BigQueryStorage
import akka.stream.scaladsl.Source
import com.google.cloud.bigquery.storage.v1.stream.ReadSession.TableReadOptions
import scala.concurrent.Future

val sourceOfSources: Source[(ReadSession.Schema, Seq[Source[ReadRowsResponse.Rows, NotUsed]]), Future[NotUsed]] =
  BigQueryStorage.create("projectId", "datasetId", "tableId", DataFormat.AVRO)
Java
sourceimport akka.stream.alpakka.googlecloud.bigquery.storage.BigQueryRecord;
import akka.stream.alpakka.googlecloud.bigquery.storage.BigQueryStorageSettings;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryArrowStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryAvroStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.BigQueryStorageAttributes;
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.GrpcBigQueryStorageReader;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import scala.Tuple2;
import com.google.cloud.bigquery.storage.v1.DataFormat;
import com.google.cloud.bigquery.storage.v1.ReadSession;
import com.google.cloud.bigquery.storage.v1.storage.ReadRowsResponse;
import akka.http.javadsl.unmarshalling.Unmarshaller;

Source<
        Tuple2<
            com.google.cloud.bigquery.storage.v1.stream.ReadSession.Schema,
            List<Source<ReadRowsResponse.Rows, NotUsed>>>,
        CompletionStage<NotUsed>>
    sourceOfSources =
        BigQueryStorage.create("projectId", "datasetId", "tableId", DataFormat.AVRO);

Secondly, by specifying TableReadOptions, we can narrow down the amount of data returned, filtering down the columns returned, and/or a row_restriction. This is defined as:

SQL text filtering statement, similar to a WHERE clause in a query. Currently, only a single predicate that is a comparison between a column and a constant value is supported. Aggregates are not supported.

Scala
sourceval readOptions = TableReadOptions(selectedFields = Seq("stringField", "intField"), rowRestriction = "intField >= 5")
val sourceOfSourcesFiltered
    : Source[(ReadSession.Schema, Seq[Source[ReadRowsResponse.Rows, NotUsed]]), Future[NotUsed]] =
  BigQueryStorage.create("projectId", "datasetId", "tableId", DataFormat.AVRO, Some(readOptions))
Java
sourceReadSession.TableReadOptions readOptions =
    ReadSession.TableReadOptions.newBuilder()
        .setSelectedFields(0, "stringField")
        .setSelectedFields(1, "intField")
        .setRowRestriction("intField >= 5")
        .build();

Source<
        Tuple2<
            com.google.cloud.bigquery.storage.v1.stream.ReadSession.Schema,
            List<Source<ReadRowsResponse.Rows, NotUsed>>>,
        CompletionStage<NotUsed>>
    sourceOfSourcesFiltered =
        BigQueryStorage.create(
            "projectId", "datasetId", "tableId", DataFormat.AVRO, readOptions, 1);

You can then choose to read and process these streams as is or merged. You can process the streams merged in rows. You need to provide a ByteString Unmarshaller based on the format requested.

Scala
sourceimplicit val unmarshaller: FromByteStringUnmarshaller[List[BigQueryRecord]] =
  mock[FromByteStringUnmarshaller[List[BigQueryRecord]]]
val sequentialSource: Source[List[BigQueryRecord], Future[NotUsed]] =
  BigQueryStorage.createMergedStreams("projectId", "datasetId", "tableId", DataFormat.AVRO)
Java
sourceUnmarshaller<ByteString, List<BigQueryRecord>> unmarshaller = null;
Source<List<BigQueryRecord>, CompletionStage<NotUsed>> sequentialSource =
    BigQueryStorage.<List<BigQueryRecord>>createMergedStreams(
        "projectId", "datasetId", "tableId", DataFormat.AVRO, unmarshaller);

Or process the stream of rows individually:

Scala
sourceimport akka.NotUsed
import com.google.cloud.bigquery.storage.v1.storage.ReadRowsResponse
import com.google.cloud.bigquery.storage.v1.DataFormat
import com.google.cloud.bigquery.storage.v1.stream.ReadSession
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.BigQueryStorage
import akka.stream.scaladsl.Source
import com.google.cloud.bigquery.storage.v1.stream.ReadSession.TableReadOptions
import scala.concurrent.Future

val sourceOfSources: Source[(ReadSession.Schema, Seq[Source[ReadRowsResponse.Rows, NotUsed]]), Future[NotUsed]] =
  BigQueryStorage.create("projectId", "datasetId", "tableId", DataFormat.AVRO)
Java
sourceimport akka.stream.alpakka.googlecloud.bigquery.storage.BigQueryRecord;
import akka.stream.alpakka.googlecloud.bigquery.storage.BigQueryStorageSettings;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryArrowStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryAvroStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.javadsl.BigQueryStorage;
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.BigQueryStorageAttributes;
import akka.stream.alpakka.googlecloud.bigquery.storage.scaladsl.GrpcBigQueryStorageReader;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import scala.Tuple2;
import com.google.cloud.bigquery.storage.v1.DataFormat;
import com.google.cloud.bigquery.storage.v1.ReadSession;
import com.google.cloud.bigquery.storage.v1.storage.ReadRowsResponse;
import akka.http.javadsl.unmarshalling.Unmarshaller;

Source<
        Tuple2<
            com.google.cloud.bigquery.storage.v1.stream.ReadSession.Schema,
            List<Source<ReadRowsResponse.Rows, NotUsed>>>,
        CompletionStage<NotUsed>>
    sourceOfSources =
        BigQueryStorage.create("projectId", "datasetId", "tableId", DataFormat.AVRO);

Since Avro and Arrow are the formats available, streams for those specific formats can be created.

You can read Arrow Record streams merged

Scala
sourceval arrowSequentialSource: Source[Seq[BigQueryRecord], Future[NotUsed]] =
  BigQueryArrowStorage.readRecordsMerged("projectId", "datasetId", "tableId")
Java
sourceSource<List<BigQueryRecord>, CompletionStage<NotUsed>> arrowSequentialSource =
    BigQueryArrowStorage.readRecordsMerged("projectId", "datasetId", "tableId");

You can read Arrow Record streams individually

Scala
sourceval arrowParallelSource: Source[Seq[Source[BigQueryRecord, NotUsed]], Future[NotUsed]] =
  BigQueryArrowStorage.readRecords("projectId", "datasetId", "tableId")
Java
sourceSource<List<Source<BigQueryRecord, NotUsed>>, CompletionStage<NotUsed>> arrowParallelSource =
    BigQueryArrowStorage.readRecords("projectId", "datasetId", "tableId");

You can read Avro Record streams merged

Scala
sourceval avroSequentialSource: Source[Seq[BigQueryRecord], Future[NotUsed]] =
  BigQueryAvroStorage.readRecordsMerged("projectId", "datasetId", "tableId")
Java
sourceSource<List<BigQueryRecord>, CompletionStage<NotUsed>> avroSequentialSource =
    BigQueryAvroStorage.readRecordsMerged("projectId", "datasetId", "tableId");

You can read Avro Record streams individually

Scala
sourceval avroParallelSource: Source[Seq[Source[BigQueryRecord, NotUsed]], Future[NotUsed]] =
  BigQueryAvroStorage.readRecords("projectId", "datasetId", "tableId")
Java
sourceSource<List<Source<BigQueryRecord, NotUsed>>, CompletionStage<NotUsed>> avroParallelSource =
    BigQueryAvroStorage.readRecords("projectId", "datasetId", "tableId");

Running the test code

The tests use a BigQueryMockServer that implements the server defined in the protobuf for the Storage API. It essentially provides a mock table on which to query. Tests can be started from sbt by running:

sbt
> google-cloud-bigquery-storage/test
Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.