Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add BigQuery Loader V2 #877

Merged
merged 4 commits into from
Feb 18, 2025
Merged

Add BigQuery Loader V2 #877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bc5d248
Add BQ Loader v2 info
mscwilson Feb 4, 2025
ce11b44
Restore previous versions image
mscwilson Feb 4, 2025
71c397e
Amendments to BigQuery Loader v2 to reflect Q4 2024 updates (#1142)
istreeter Feb 10, 2025
ab47a9a
Update config reference
benjben Feb 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/api-reference/loaders-storage-targets/bigquery-loader/_cross-cloud-diagram.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
```mdx-code-block
import Admonition from '@theme/Admonition';
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
import LoaderDiagram from '@site/docs/api-reference/loaders-storage-targets/bigquery-loader/_diagram.md';
```

<Tabs groupId="cloud" queryString lazy>
<TabItem value="aws" label="AWS">
<LoaderDiagram {...props} stream="Kinesis" cloud="AWS"/>
</TabItem>
<TabItem value="gcp" label="GCP">
<LoaderDiagram {...props} stream="Pub/Sub" cloud="GCP"/>
</TabItem>
<TabItem value="azure" label="Azure">
<LoaderDiagram {...props} stream="Kafka" cloud="Azure"/>
</TabItem>
</Tabs>
22 changes: 22 additions & 0 deletions docs/api-reference/loaders-storage-targets/bigquery-loader/_deploy-overview.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
```mdx-code-block
import {versions} from '@site/src/componentVersions';
import CodeBlock from '@theme/CodeBlock';
```

<p>The BigQuery Loader is published as a Docker image which you can run on any {props.cloud} VM.</p>

<CodeBlock language="bash">{
`docker pull snowplow/bigquery-loader-${props.stream}:${versions.bqLoader}
`}</CodeBlock>

To run the loader, mount your config file into the docker image, and then provide the file path on the command line.

<CodeBlock language="bash">{
`docker run \\
--mount=type=bind,source=/path/to/myconfig,destination=/myconfig \\
snowplow/bigquery-loader-${props.stream}:${versions.bqLoader} \\
--config=/myconfig/loader.hocon \\
--iglu-config /myconfig/iglu.hocon
`}</CodeBlock>

Where `loader.hocon` is loader's [configuration file](/docs/api-reference/loaders-storage-targets/bigquery-loader/#configuring-the-loader) and `iglu.hocon` is [iglu resolver](/docs/api-reference/iglu/iglu-resolver/index.md) configuration.
31 changes: 11 additions & 20 deletions docs/api-reference/loaders-storage-targets/bigquery-loader/_diagram.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,15 @@
At the high level, BigQuery loader reads enriched Snowplow events in real time and loads them in BigQuery using the Storage Write API.
```mdx-code-block
import Mermaid from '@theme/Mermaid';
import Link from '@docusaurus/Link';
```
<p>The BigQuery Streaming Loader on {props.cloud} is a fully streaming application that continually pulls events from {props.stream} and writes to BigQuery using the <Link to="https://cloud.google.com/bigquery/docs/write-api">BigQuery Storage API</Link>.</p>

```mermaid
<Mermaid value={`
flowchart LR
stream[["<b>Enriched events</b>\n(Pub/Sub stream)"]]
loader{{"<b>BigQuery Loader</b>\n(Loader, Mutator and Repeater apps)"}}
subgraph BigQuery
stream[["<b>Enriched Events</b>\n(${props.stream} stream)"]]
loader{{"<b>BigQuery Loader</b>"}}
subgraph bigquery [BigQuery]
table[("<b>Events table</b>")]
end
stream-->loader-->BigQuery
```

BigQuery loader consists of three applications: Loader, Mutator and Repeater. The following diagram illustrates the interaction between them and BigQuery:

```mermaid
sequenceDiagram
loop
Note over Loader: Read a small batch of events
Loader-->>+Mutator: Communicate event types (via Pub/Sub)
Loader->>BigQuery: Send events using the Storage Write API
Mutator-->>-BigQuery: Adjust column types if necessary
Repeater->>BigQuery: Resend events that failed<br/>because columns were not up to date
end
```
stream-->loader-->|BigQuery Storage API|bigquery
`}/>
16 changes: 16 additions & 0 deletions ...ers-storage-targets/bigquery-loader/configuration-reference/_bigquery_config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
<tr>
<td><code>output.good.project</code></td>
<td>Required. The GCP project to which the BigQuery dataset belongs</td>
</tr>
<tr>
<td><code>output.good.dataset</code></td>
<td>Required. The BigQuery dataset to which events will be loaded</td>
</tr>
<tr>
<td><code>output.good.table</code></td>
<td>Optional. Default value <code>events</code>. Name to use for the events table</td>
</tr>
<tr>
<td><code>output.good.credentials</code></td>
<td>Optional. Service account credentials (JSON). If not set, default credentials will be sourced from the usual locations, e.g. file pointed to by the <code>GOOGLE_APPLICATION_CREDENTIALS</code> environment variable </td>
</tr>
139 changes: 139 additions & 0 deletions ...aders-storage-targets/bigquery-loader/configuration-reference/_common_config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
```mdx-code-block
import Link from '@docusaurus/Link';
```

<tr>
<td><code>batching.maxBytes</code></td>
<td>Optional. Default value <code>10000000</code>. Events are emitted to BigQuery when the batch reaches this size in bytes</td>
</tr>
<tr>
<td><code>batching.maxDelay</code></td>
<td>Optional. Default value <code>1 second</code>. Events are emitted to BigQuery after a maximum of this duration, even if the <code>maxBytes</code> size has not been reached</td>
</tr>
<tr>
<td><code>batching.writeBatchConcurrency</code></td>
<td>Optional. Default value 2. How many batches can we send simultaneously over the network to BigQuery</td>
</tr>
<tr>
<td><code>cpuParallelism.parseBytesFactor</code></td>
<td>
Optional. Default value <code>0.1</code>.
Controls how many batches of bytes we can parse into enriched events simultaneously.
E.g. If there are 2 cores and <code>parseBytesFactor = 0.1</code> then only one batch gets processed at a time.
Adjusting this value can cause the app to use more or less of the available CPU.
</td>
</tr>
<tr>
<td><code>cpuParallelism.transformFactor</code></td>
<td>
Optional. Default value <code>0.75</code>.
Controls how many batches of enriched events we can transform into BigQuery format simultaneously.
E.g. If there are 4 cores and <code>transformFactor = 0.75</code> then 3 batches gets processed in parallel.
Adjusting this value can cause the app to use more or less of the available CPU.
</td>
</tr>
<tr>
<td><code>retries.setupErrors.delay</code></td>
<td>
Optional. Default value <code>30 seconds</code>.
Configures exponential backoff on errors related to how BigQuery is set up for this loader.
Examples include authentication errors and permissions errors.
This class of errors are reported periodically to the monitoring webhook.
</td>
</tr>
<tr>
<td><code>retries.transientErrors.delay</code></td>
<td>
Optional. Default value <code>1 second</code>.
Configures exponential backoff on errors that are likely to be transient.
Examples include server errors and network errors.
</td>
</tr>
<tr>
<td><code>retries.transientErrors.attempts</code></td>
<td>Optional. Default value 5. Maximum number of attempts to make before giving up on a transient error.</td>
</tr>
<tr>
<td><code>skipSchemas</code></td>
<td>
Optional, e.g. <code>["iglu:com.example/skipped1/jsonschema/1-0-0"]</code> or with wildcards <code>["iglu:com.example/skipped2/jsonschema/1-*-*"]</code>.
A list of schemas that won't be loaded to BigQuery.
This feature could be helpful when recovering from edge-case schemas which for some reason cannot be loaded to the table.
</td>
</tr>
<tr>
<td><code>legacyColumnMode</code></td>
<td>Optional. Default value <code>false</code>.
When this mode is enabled, the loader uses the legacy column style used by the v1 BigQuery loader.
For example, an entity for a <code>1-0-0</code> schema is loaded into a column ending in <code>_1_0_0</code>, instead of a column ending in <code>_1</code>.
This feature could be helpful when migrating from the v1 loader to the v2 loader.
</td>
</tr>
<tr>
<td><code>legacyColumns</code></td>
<td>
Optional, e.g. <code>["iglu:com.example/legacy/jsonschema/1-0-0"]</code> or with wildcards <code>["iglu:com.example/legacy/jsonschema/1-*-*"]</code>.
Schemas for which to use the legacy column style used by the v1 BigQuery loader, even when <code>legacyColumnMode</code> is disabled.
</td>
</tr>
<tr>
<td><code>exitOnMissingIgluSchema</code></td>
<td>
Optional. Default value <code>true</code>.
Whether the loader should crash and exit if it fails to resolve an Iglu Schema.
We recommend `true` because Snowplow enriched events have already passed validation, so a missing schema normally indicates an error that needs addressing.
Change to <code>false</code> so events go the failed events stream instead of crashing the loader.
</td>
</tr>
<tr>
<td><code>monitoring.metrics.statsd.hostname</code></td>
<td>Optional. If set, the loader sends statsd metrics over UDP to a server on this host name.</td>
</tr>
<tr>
<td><code>monitoring.metrics.statsd.port</code></td>
<td>Optional. Default value 8125. If the statsd server is configured, this UDP port is used for sending metrics.</td>
</tr>
<tr>
<td><code>monitoring.metrics.statsd.tags.*</code></td>
<td>Optional. A map of key/value pairs to be sent along with the statsd metric.</td>
</tr>
<tr>
<td><code>monitoring.metrics.statsd.period</code></td>
<td>Optional. Default <code>1 minute</code>. How often to report metrics to statsd.</td>
</tr>
<tr>
<td><code>monitoring.metrics.statsd.prefix</code></td>
<td>Optional. Default <code>snowplow.bigquery-loader</code>. Prefix used for the metric name when sending to statsd.</td>
</tr>
<tr>
<td><code>monitoring.webhook.endpoint</code></td>
<td>Optional, e.g. <code>https://webhook.example.com</code>. The loader will send to the webhook a payload containing details of any error related to how BigQuery is set up for this loader.</td>
</tr>
<tr>
<td><code>monitoring.webhook.tags.*</code></td>
<td>Optional. A map of key/value strings to be included in the payload content sent to the webhook.</td>
</tr>
<tr>
<td><code>monitoring.webhook.heartbeat.*</code></td>
<td>Optional. Default value <code>5.minutes</code>. How often to send a heartbeat event to the webhook when healthy.</td>
</tr>
<tr>
<td><code>monitoring.sentry.dsn</code></td>
<td>Optional. Set to a Sentry URI to report unexpected runtime exceptions.</td>
</tr>
<tr>
<td><code>monitoring.sentry.tags.*</code></td>
<td>Optional. A map of key/value strings which are passed as tags when reporting exceptions to Sentry.</td>
</tr>
<tr>
<td><code>telemetry.disable</code></td>
<td>Optional. Set to <code>true</code> to disable <Link to="/docs/getting-started-on-community-edition/telemetry/">telemetry</Link>.</td>
</tr>
<tr>
<td><code>telemetry.userProvidedId</code></td>
<td>Optional. See <Link to="/docs/getting-started-on-community-edition/telemetry/#how-can-i-help">here</Link> for more information.</td>
</tr>
<tr>
<td><code>http.client.maxConnectionsPerServer</code></td>
<td> Optional. Default value 4. Configures the internal HTTP client used for iglu resolver, alerts and telemetry. The maximum number of open HTTP requests to any single server at any one time.</td>
</tr>
28 changes: 28 additions & 0 deletions ...oaders-storage-targets/bigquery-loader/configuration-reference/_kafka_config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
<tr>
<td><code>input.topicName</code></td>
<td>Required. Name of the Kafka topic for the source of enriched events.</td>
</tr>
<tr>
<td><code>input.bootstrapServers</code></td>
<td>Required. Hostname and port of Kafka bootstrap servers hosting the source of enriched events.</td>
</tr>
<tr>
<td><code>input.consumerConf.*</code></td>
<td>Optional. A map of key/value pairs for <a href="https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html" target="_blank">any standard Kafka consumer configuration option</a>.</td>
</tr>
<tr>
<td><code>output.bad.topicName</code></td>
<td>Required. Name of the Kafka topic that will receive failed events.</td>
</tr>
<tr>
<td><code>output.bad.bootstrapServers</code></td>
<td>Required. Hostname and port of Kafka bootstrap servers hosting the bad topic</td>
</tr>
<tr>
<td><code>output.bad.producerConf.*</code></td>
<td>Optional. A map of key/value pairs for <a href="https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html" target="_blank">any standard Kafka producer configuration option</a>.</td>
</tr>
<tr>
<td><code>output.bad.maxRecordSize.*</code></td>
<td>Optional. Default value 1000000. Any single failed event sent to Kafka should not exceed this size in bytes</td>
</tr>
68 changes: 68 additions & 0 deletions ...ders-storage-targets/bigquery-loader/configuration-reference/_kinesis_config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
<tr>
<td><code>input.streamName</code></td>
<td>Required. Name of the Kinesis stream with the enriched events</td>
</tr>
<tr>
<td><code>input.appName</code></td>
<td>Optional, default <code>snowplow-bigquery-loader</code>. Name to use for the dynamodb table, used by the underlying Kinesis Consumer Library for managing leases.</td>
</tr>
<tr>
<td><code>input.initialPosition</code></td>
<td>Optional, default <code>LATEST</code>. Allowed values are <code>LATEST</code>, <code>TRIM_HORIZON</code>, <code>AT_TIMESTAMP</code>. When the loader is deployed for the first time, this controls from where in the kinesis stream it should start consuming events. On all subsequent deployments of the loader, the loader will resume from the offsets stored in the DynamoDB table.</td>
</tr>
<tr>
<td><code>input.initialPosition.timestamp</code></td>
<td>Required if <code>input.initialPosition</code> is <code>AT_TIMESTAMP</code>. A timestamp in ISO8601 format from where the loader should start consuming events.</td>
</tr>
<tr>
<td><code>input.retrievalMode</code></td>
<td>Optional, default Polling. Change to FanOut to enable the enhance fan-out feature of Kinesis.</td>
</tr>
<tr>
<td><code>input.retrievalMode.maxRecords</code></td>
<td>Optional. Default value 1000. How many events the Kinesis client may fetch in a single poll. Only used when `input.retrievalMode` is Polling.</td>
</tr>
<tr>
<td><code>input.workerIdentifier</code></td>
<td>Optional. Defaults to the <code>HOSTNAME</code> environment variable. The name of this KCL worker used in the dynamodb lease table.</td>
</tr>
<tr>
<td><code>input.leaseDuration</code></td>
<td>Optional. Default value <code>10 seconds</code>. The duration of shard leases. KCL workers must periodically refresh leases in the dynamodb table before this duration expires.</td>
</tr>
<tr>
<td><code>input.maxLeasesToStealAtOneTimeFactor</code></td>
<td>Optional. Default value <code>2.0</code>. Controls how to pick the max number of shard leases to steal at one time. E.g. If there are 4 available processors, and <code>maxLeasesToStealAtOneTimeFactor = 2.0</code>, then allow the loader to steal up to 8 leases. Allows bigger instances to more quickly acquire the shard-leases they need to combat latency.</td>
</tr>
<tr>
<td><code>input.checkpointThrottledBackoffPolicy.minBackoff</code></td>
<td>Optional. Default value <code>100 milliseconds</code>. Initial backoff used to retry checkpointing if we exceed the DynamoDB provisioned write limits.</td>
</tr>
<tr>
<td><code>input.checkpointThrottledBackoffPolicy.maxBackoff</code></td>
<td>Optional. Default value <code>1 second</code>. Maximum backoff used to retry checkpointing if we exceed the DynamoDB provisioned write limits.</td>
</tr>
<tr>
<td><code>output.bad.streamName</code></td>
<td>Required. Name of the Kinesis stream that will receive failed events.</td>
</tr>
<tr>
<td><code>output.bad.throttledBackoffPolicy.minBackoff</code></td>
<td>Optional. Default value <code>100 milliseconds</code>. Initial backoff used to retry sending failed events if we exceed the Kinesis write throughput limits.</td>
</tr>
<tr>
<td><code>output.bad.throttledBackoffPolicy.maxBackoff</code></td>
<td>Optional. Default value <code>1 second</code>. Maximum backoff used to retry sending failed events if we exceed the Kinesis write throughput limits.</td>
</tr>
<tr>
<td><code>output.bad.recordLimit</code></td>
<td>Optional. Default value 500. The maximum number of records we are allowed to send to Kinesis in 1 PutRecords request.</td>
</tr>
<tr>
<td><code>output.bad.byteLimit</code></td>
<td>Optional. Default value 5242880. The maximum number of bytes we are allowed to send to Kinesis in 1 PutRecords request.</td>
</tr>
<tr>
<td><code>output.bad.maxRecordSize.*</code></td>
<td>Optional. Default value 1000000. Any single event failed event sent to Kinesis should not exceed this size in bytes</td>
</tr>
49 changes: 49 additions & 0 deletions ...aders-storage-targets/bigquery-loader/configuration-reference/_pubsub_config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
<tr>
<td><code>input.subscription</code></td>
<td>Required, e.g. <code>projects/myproject/subscriptions/snowplow-enriched</code>. Name of the Pub/Sub subscription with the enriched events</td>
</tr>
<tr>
<td><code>input.parallelPullFactor</code></td>
<td>Optional. Default value 0.5. <code>parallelPullFactor * cpu count</code> will determine the number of threads used internally by the Pub/Sub client library for fetching events</td>
</tr>
<tr>
<td><code>input.durationPerAckExtension</code></td>
<td>Optional. Default value <code>60 seconds</code>. Pubsub ack deadlines are extended for this duration when needed.</td>
</tr>
<tr>
<td><code>input.minRemainingAckDeadline</code></td>
<td>
Optional. Default value 0.1.
Controls when ack deadlines are re-extended, for a message that is close to exceeding its ack deadline.
For example, if <code>durationPerAckExtension</code> is <code>60 seconds</code> and <code>minRemainingAckDeadline</code> is <code>0.1</code> then the loader
will wait until there is <code>6 seconds</code> left of the remining deadline, before re-extending the message deadline.
</td>
</tr>
<tr>
<td><code>input.maxMessagesPerPull</code></td>
<td>Optional. Default value 1000. How many Pub/Sub messages to pull from the server in a single request.</td>
</tr>
<tr>
<td><code>input.debounceRequests</code></td>
<td>
Optional. Default value <code>100 millis</code>.
Adds an artifical delay between consecutive requests to Pub/Sub for more messages.
Under some circumstances, this was found to slightly alleviate a problem in which Pub/Sub might re-deliver the same messages multiple times.
</td>
</tr>
<tr>
<td><code>output.bad.topic</code></td>
<td>Required, e.g. <code>projects/myproject/topics/snowplow-bad</code>. Name of the Pub/Sub topic that will receive failed events.</td>
</tr>
<tr>
<td><code>output.bad.batchSize</code></td>
<td>Optional. Default value 1000. Bad events are sent to Pub/Sub in batches not exceeding this count.</td>
</tr>
<tr>
<td><code>output.bad.requestByteThreshold</code></td>
<td>Optional. Default value 1000000. Bad events are sent to Pub/Sub in batches with a total size not exceeding this byte threshold</td>
</tr>
<tr>
<td><code>output.bad.maxRecordSize</code></td>
<td>Optional. Default value 10000000. Any single failed event sent to Pub/Sub should not exceed this size in bytes</td>
</tr>
Loading