Package io.github.flink.gcp.connector.bigtable.source

Class BigtableSourceBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.bigtable.source.BigtableSourceBuilder<T>
Type Parameters:
T - the record type produced
@Public public class BigtableSourceBuilder<T> extends Object
Builds a BigtableSource.

Every range this builder is given is copied on the way in and validated at build(), so a configuration mistake fails where the job is assembled rather than on a TaskManager once rows begin to flow.

  • Field Details

    • DEFAULT_MAX_ROWS_PER_FETCH

      public static final int DEFAULT_MAX_ROWS_PER_FETCH
      The default maximum number of rows one fetch hands to Flink's element queue.
      See Also:

    • DEFAULT_MAX_BYTES_PER_FETCH

      public static final long DEFAULT_MAX_BYTES_PER_FETCH
      The default target maximum estimated bytes one fetch hands to Flink's element queue.
      See Also:

  • Method Details

    • table

      public BigtableSourceBuilder<T> table(TableDestination table)
      Sets the table to read. Required.
      Parameters:
      table - the table
      Returns:
      this builder

    • deserializer

      public BigtableSourceBuilder<T> deserializer(BigtableRowDeserializationSchema<T> deserializer)
      Sets the deserializer turning rows into records. Required.
      Parameters:
      deserializer - the deserializer
      Returns:
      this builder

    • rowRange

      public BigtableSourceBuilder<T> rowRange(com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange range)
      Adds a row-key range to read. Repeatable; with no range set at all the whole table is read.

      Overlapping ranges are merged at build() rather than rejected — nested prefixes are easy to configure by accident — but an empty range is rejected, because a range that reads nothing under a successful job looks exactly like a job with nothing to read.

      Parameters:
      range - the range; copied, so later changes to it do not affect the source
      Returns:
      this builder

    • rowRange

      public BigtableSourceBuilder<T> rowRange(com.google.protobuf.ByteString startClosed, com.google.protobuf.ByteString endOpen)
      Adds a row-key range to read, from an inclusive start to an exclusive end.
      Parameters:
      startClosed - the first row key to read
      endOpen - the first row key past the range
      Returns:
      this builder

    • rowRange

      public BigtableSourceBuilder<T> rowRange(String startClosed, String endOpen)
      Adds a row-key range to read, from an inclusive start to an exclusive end, given as UTF-8 text.
      Parameters:
      startClosed - the first row key to read
      endOpen - the first row key past the range
      Returns:
      this builder

    • prefix

      public BigtableSourceBuilder<T> prefix(com.google.protobuf.ByteString prefix)
      Adds every row whose key starts with a prefix. Repeatable, and sugar for the range that prefix describes.

      The conversion is the client library's, which handles the two cases a hand-rolled one gets wrong: a prefix that is all 0xFF bytes has no successor and becomes a range running to the end of the table, and a prefix ending in 0xFF bytes carries the increment into an earlier byte. An empty prefix is the whole table.

      Parameters:
      prefix - the row-key prefix
      Returns:
      this builder

    • prefix

      public BigtableSourceBuilder<T> prefix(String prefix)
      Adds every row whose key starts with a prefix, given as UTF-8 text.
      Parameters:
      prefix - the row-key prefix
      Returns:
      this builder

    • filter

      public BigtableSourceBuilder<T> filter(com.google.cloud.bigtable.data.v2.models.Filters.Filter filter)
      Sets the server-side filter every read applies. Optional; last writer wins.

      One filter, applied identically to every split, which is safe by construction: Bigtable's filter language has no row-count limiter — its limit and offset filters count cells within a row — so nothing expressible here can depend on how the key space was divided.

      Per-cell shaping is all expressible through a filter: which families and qualifiers to return, which timestamp window, how many versions of a cell. There are no separate knobs for those, and a filter is also the cheapest thing a scan can carry, since what it excludes never leaves the server.

      Parameters:
      filter - the filter, built through the client's Filters factory
      Returns:
      this builder

    • appProfileId

      public BigtableSourceBuilder<T> appProfileId(String appProfileId)
      Sets the application profile every call routes through. Optional; the instance's default profile is used when this is not set.

      It is a source option rather than part of TableDestination because it chooses a path to the data, not the data's address. A Data Boost profile is named here like any other.

      Parameters:
      appProfileId - the application profile id
      Returns:
      this builder

    • serviceAccountKeyFile

      public BigtableSourceBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Authenticates the source with the service-account JSON key at the given path instead of application-default credentials. The JobManager reads the file when its enumerator is created or restored, whether or not that enumerator goes on to sample row keys, and each TaskManager reads it when its reader is created, before any split is assigned to it. Every eligible process must therefore see the same path. Optional; when unset the source uses application-default credentials.

      Service-account keys are long-lived secrets. Prefer an attached service account or Workload Identity where the deployment supports one. This setting cannot be combined with emulatorEndpoint(String), whose plaintext channel carries no credentials.

      Parameters:
      serviceAccountKeyFile - the service-account JSON key-file path
      Returns:
      this builder

    • emulatorEndpoint

      public BigtableSourceBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the source at an emulator, over a plaintext channel with no credentials. Never production.
      Parameters:
      emulatorEndpoint - the emulator endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • maxRowsPerFetch

      public BigtableSourceBuilder<T> maxRowsPerFetch(int maxRowsPerFetch)
      Sets the maximum number of input rows one fetch hands to Flink's element queue. Optional; defaults to 1000.

      The fetch returns when either this row limit or maxBytesPerFetch(long) is reached. Lower values reduce the source reader's queued memory and the delay before a checkpoint or cancellation can be observed, at the cost of more fetch hand-offs.

      Parameters:
      maxRowsPerFetch - the positive maximum number of rows per fetch
      Returns:
      this builder

    • maxBytesPerFetch

      public BigtableSourceBuilder<T> maxBytesPerFetch(long maxBytesPerFetch)
      Sets the target maximum estimated bytes one fetch hands to Flink's element queue. Optional; defaults to 8388608L bytes.

      The estimate covers the decoded row key, cell values, and cell metadata measured while the SDK materialises each row. The next row is held for the following fetch when adding it would exceed this target. One row larger than the target is still handed over alone so the source can make progress.

      The fetch returns when either this byte target or maxRowsPerFetch(int) is reached. Lower values reduce the source reader's queued memory, while higher values can reduce fetch hand-offs for wide rows.

      Parameters:
      maxBytesPerFetch - the positive target maximum estimated bytes per fetch
      Returns:
      this builder

    • build

      public org.apache.flink.api.connector.source.Source<T,RowRangeSplit,BigtableScanEnumeratorState> build()
      Builds the source.
      Returns:
      the source
      Throws:
      IllegalStateException - if a required option was not set
      IllegalArgumentException - if a configured range is empty, or if the filter is larger than the service accepts