Package io.github.flink.gcp.connector.bigquery.sink

Class BigQuerySinkBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.bigquery.sink.BigQuerySinkBuilder<T>
Type Parameters:
T - type of the records written by the sink
@Public public class BigQuerySinkBuilder<T> extends Object
Builder for BigQuery sinks, obtained from BigQuerySink.builder().

Required settings: a serializer and a destination. The destination is set through either table(TableDestination) (fixed table) or destinationResolver(DestinationResolver) (per-record dynamic destinations); the two override each other and the last call wins.

  • Method Details

    • writeMethod

      public BigQuerySinkBuilder<T> writeMethod(WriteMethod writeMethod)
      Sets the write method. Defaults to WriteMethod.STORAGE_API_AT_LEAST_ONCE.
      Parameters:
      writeMethod - the write method
      Returns:
      this builder

    • table

      public BigQuerySinkBuilder<T> table(TableDestination table)
      Writes every record to the given fixed table. Overrides any previously set table or destination resolver.
      Parameters:
      table - the destination table
      Returns:
      this builder

    • destinationResolver

      public BigQuerySinkBuilder<T> destinationResolver(DestinationResolver<? super T> destinationResolver)
      Resolves the destination table per record (dynamic destinations). Overrides any previously set table or destination resolver.
      Parameters:
      destinationResolver - the resolver
      Returns:
      this builder

    • serializer

      public BigQuerySinkBuilder<T> serializer(BigQueryProtoSerializationSchema<? super T> serializer)
      Sets the record serializer. Required. Besides the row bytes, it supplies each destination's table schema, and returning null from it skips a record rather than failing it.
      Parameters:
      serializer - the serializer
      Returns:
      this builder

    • additionalFields

      public BigQuerySinkBuilder<T> additionalFields(AdditionalFields<? super T> additionalFields)
      Appends physical fields derived from each non-skipped input record.

      The fields are added to protobuf rows and to the physical BigQuery schema used by every write method for table creation and schema reconciliation. When this method is not called, it adds no fields or provider calls. If no other row decorator such as cdcOptions(CdcOptions) is configured, the sink uses the serializer's schema, descriptor, and row bytes unchanged.

      Parameters:
      additionalFields - ordered physical fields to append
      Returns:
      this builder

    • cdcOptions

      public BigQuerySinkBuilder<T> cdcOptions(CdcOptions<? super T> cdcOptions)
      Enables BigQuery change data capture for records appended through the Storage Write API default stream.

      The desired primary key and optional maximum-staleness policy are configured separately through cdcTableOptions(CdcTableOptions) or cdcTableOptionsProvider(CdcTableOptionsProvider). The sink creates a missing physical schema and primary key through the Tables API only when CreateDisposition.CREATE_IF_NEEDED permits it. Existing-table handling is selected through cdcTableReconciliationPolicy(CdcTableReconciliationPolicy). CDC pseudocolumns are added to write rows only, never to the physical schema. Rejected for WriteMethod.STORAGE_API_EXACTLY_ONCE and WriteMethod.FILE_LOADS.

      Parameters:
      cdcOptions - the per-record CDC providers
      Returns:
      this builder

    • cdcTableOptions

      public BigQuerySinkBuilder<T> cdcTableOptions(CdcTableOptions cdcTableOptions)
      Applies the same desired CDC table contract to every destination.

      Primary-key columns are required only when the sink must create a missing table or when CdcTableReconciliationPolicy.RECONCILE needs an authoritative key. An existing table under CdcTableReconciliationPolicy.VERIFY_ONLY may supply its key from BigQuery metadata. Overrides any previously set options or provider.

      Parameters:
      cdcTableOptions - the desired CDC table contract
      Returns:
      this builder

    • cdcTableOptionsProvider

      public BigQuerySinkBuilder<T> cdcTableOptionsProvider(CdcTableOptionsProvider cdcTableOptionsProvider)
      Resolves the desired CDC table contract per destination. Overrides fixed CDC table options.
      Parameters:
      cdcTableOptionsProvider - the provider
      Returns:
      this builder

    • cdcTableReconciliationPolicy

      public BigQuerySinkBuilder<T> cdcTableReconciliationPolicy(CdcTableReconciliationPolicy policy)
      Sets how the sink handles a CDC destination table that already exists. Defaults to CdcTableReconciliationPolicy.VERIFY_ONLY.

      This policy is independent of CreateDisposition: reconciliation never authorizes creation under CreateDisposition.CREATE_NEVER.

      Parameters:
      policy - the existing-table policy
      Returns:
      this builder

    • createDisposition

      public BigQuerySinkBuilder<T> createDisposition(CreateDisposition createDisposition)
      Sets the table create disposition. Defaults to CreateDisposition.CREATE_IF_NEEDED.
      Parameters:
      createDisposition - the create disposition
      Returns:
      this builder

    • tableCreateOptions

      public BigQuerySinkBuilder<T> tableCreateOptions(TableCreateOptions tableCreateOptions)
      Applies the same creation options to every table created under CreateDisposition.CREATE_IF_NEEDED. Overrides any previously set options or provider. Defaults to TableCreateOptions.defaults() (plain tables). CDC primary-key and maximum-staleness properties belong to CdcTableOptions instead.
      Parameters:
      tableCreateOptions - the creation options
      Returns:
      this builder

    • tableCreateOptionsProvider

      public BigQuerySinkBuilder<T> tableCreateOptionsProvider(TableCreateOptionsProvider tableCreateOptionsProvider)
      Resolves creation options per destination for tables created under CreateDisposition.CREATE_IF_NEEDED. Overrides any previously set options or provider.
      Parameters:
      tableCreateOptionsProvider - the provider
      Returns:
      this builder

    • schemaUpdateOptions

      public BigQuerySinkBuilder<T> schemaUpdateOptions(SchemaUpdateOptions schemaUpdateOptions)
      Sets the options gating connector-driven table schema updates. Defaults to SchemaUpdateOptions.defaults() (updates disabled).

      Schema changes made externally (for example via DDL) are always picked up without a job restart; these options only control whether the sink may update destination table schemas itself when the serializer's schema evolves past the table's.

      Parameters:
      schemaUpdateOptions - the schema update options
      Returns:
      this builder

    • failureHandler

      public BigQuerySinkBuilder<T> failureHandler(FailureHandler<? super BigQueryFailure> failureHandler)
      Sets the policy for records that explicitly fail destination resolution or terminally fail after routing (rows rejected by the Storage Write API with per-row error details, rows that fail serialization, and rows exceeding the per-row size limit). Defaults to FailureHandler.failJob().

      The handler decides per failure: returning normally drops the record, throwing fails the write or checkpoint. Transient append failures are retried without involving the handler, and terminal request failures such as INVALID_ARGUMENT always fail the job. The sink drives the handler's lifecycle (open/flush/close) as documented on FailureHandler. The parameter is contravariant, so a cross-connector FailureHandler<FailedElement> is accepted as-is.

      Parameters:
      failureHandler - the handler
      Returns:
      this builder

    • location

      public BigQuerySinkBuilder<T> location(String location)
      Sets the BigQuery location (for example US or asia-northeast1) shared by the destination tables. Optional; setting it avoids a per-table metadata lookup when opening Storage Write API connections, and under WriteMethod.FILE_LOADS it becomes the location every load job runs in and is looked up under. When unset, FILE_LOADS derives each job's location from its destination dataset's metadata instead — one datasets.get per dataset per committer, which needs the bigquery.datasets.get permission there — so a sink routing to datasets in several regions should leave it unset.
      Parameters:
      location - the BigQuery location
      Returns:
      this builder

    • fileLoadsOptions

      public BigQuerySinkBuilder<T> fileLoadsOptions(FileLoadsOptions fileLoadsOptions)
      Sets the options specific to WriteMethod.FILE_LOADS. Required for that write method and rejected for every other one.
      Parameters:
      fileLoadsOptions - the file-loads options
      Returns:
      this builder

    • bufferedStreamOptions

      public BigQuerySinkBuilder<T> bufferedStreamOptions(BufferedStreamOptions bufferedStreamOptions)
      Sets the options specific to WriteMethod.STORAGE_API_EXACTLY_ONCE. Required for that write method (all knobs are defaulted, so BufferedStreamOptions.builder().build() is a valid value) and rejected for every other one.
      Parameters:
      bufferedStreamOptions - the buffered-stream options
      Returns:
      this builder

    • defaultStreamOptions

      public BigQuerySinkBuilder<T> defaultStreamOptions(DefaultStreamOptions defaultStreamOptions)
      Sets the options specific to WriteMethod.STORAGE_API_AT_LEAST_ONCE. Optional for that write method — the default write method is chosen by not choosing, so unlike the other write-method option objects nothing forces this one into view, and an unconfigured sink uses DefaultStreamOptions.builder().build() — and rejected for every other one.
      Parameters:
      defaultStreamOptions - the default-stream options
      Returns:
      this builder

    • serviceAccountKeyFile

      public BigQuerySinkBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Uses the service account in the given JSON key file for every BigQuery client the sink opens and, under WriteMethod.FILE_LOADS, for Cloud Storage staging as well. Optional; when unset, clients use application-default credentials.

      The builder stores only the path in the job graph. Each runtime component reads the file when it first creates a client, so the file must be available at the same path on every Task Manager that runs a writer or committer.

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

    • emulatorEndpoint

      public BigQuerySinkBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the sink's Storage Write API traffic at a BigQuery emulator instead of the production service. The write stream opened for each destination connects to the given host:port over a plaintext channel with no credentials, so this must only ever be used against an emulator (for example a testcontainers goccy/bigquery-emulator). Optional; when unset the sink connects to BigQuery with configured credentials or ADC.

      BigQuery serves its two transports on separate ports — gRPC for the Storage Write API, REST for table metadata — so this endpoint covers the gRPC half only, and a job that also creates tables, evolves their schemas or manages a CDC table needs emulatorRestEndpoint(String) beside it. That is a deviation from the sibling connectors, each of which needs one transport and so exposes one endpoint.

      Rejected under WriteMethod.FILE_LOADS: that write method stages files to Cloud Storage, which no emulator here stands in for, so an endpoint could only be half honored.

      The value is parsed here, so a malformed host:port is rejected on the client instead of surfacing as a connection failure once the job has been deployed.

      Parameters:
      emulatorEndpoint - the emulator's gRPC endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • emulatorRestEndpoint

      public BigQuerySinkBuilder<T> emulatorRestEndpoint(String emulatorRestEndpoint)
      Points the sink's table metadata traffic — table creation under CreateDisposition.CREATE_IF_NEEDED, connector-driven schema updates and the CDC table contract — at a BigQuery emulator instead of the production service. The REST client is built against http://host:port with no credentials, so this must only ever be used against an emulator. Optional; when unset the metadata client uses configured credentials or ADC — so a sink doing any of those three with only emulatorEndpoint(String) set still reaches real BigQuery.

      This is the REST half of emulatorEndpoint(String); see there for why the two are separate and for the WriteMethod.FILE_LOADS rejection, which applies to both.

      Parameters:
      emulatorRestEndpoint - the emulator's REST endpoint as host:port, without a scheme
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • build

      public org.apache.flink.api.connector.sink2.Sink<T> build()
      Builds the sink for the configured WriteMethod.
      Returns:
      the sink