You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Amar3tto (via GitHub)" <gi...@apache.org> on 2026/07/09 10:57:58 UTC
[PR] IcerbergIO Load test and benchmarking [beam]
Amar3tto opened a new pull request, #39270:
URL: https://github.com/apache/beam/pull/39270
**Successful run:** https://github.com/apache/beam/actions/runs/29006124448/job/86078012033
**Read Dataflow Job:** 2026-07-09_02_11_56-9909184907203270916
**Write Dataflow Job:** 2026-07-09_02_00_15-6048170378328054140
**Changes:**
- Moved the next classes from `it.gcp` to `it.common`:
- - `DefaultPipelineLauncher` and `AbstractPipelineLauncher`
- - `LoadTestBase`
- - Resource managers and Artifact utils
- Implemented Load test for IcebergIO Batch Read/Write
- Added IcebergIO Looker diagrams with metrics to the website
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3551067417
##########
it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java:
##########
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.it.iceberg;
+
+import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult;
+import static org.junit.Assert.assertEquals;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.text.ParseException;
+import java.time.Duration;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import org.apache.beam.it.common.PipelineLauncher;
+import org.apache.beam.it.common.PipelineOperator;
+import org.apache.beam.it.common.TestProperties;
+import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
+import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.storage.GcsResourceManager;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.io.iceberg.IcebergIO;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestPipelineOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * IcebergIO performance tests using a Hadoop catalog backed by GCS.
+ *
+ * <p>The test writes generated rows to an Iceberg table, reads the table back, validates the number
+ * of rows, and exports write and read performance metrics.
+ *
+ * <p>Example trigger command:
+ *
+ * <pre>
+ * ./gradlew :it:iceberg:integrationTest \
+ * --tests "org.apache.beam.it.iceberg.IcebergIOLT" \
+ * -Dconfiguration=local \
+ * -Dproject=[gcpProject] \
+ * -DartifactBucket=[artifactBucket]
+ * </pre>
+ */
+public final class IcebergIOLT extends IOLoadTestBase {
+
+ private static final String READ_ELEMENT_METRIC_NAME = "read_count";
+ private static final String NAMESPACE = "default";
+
+ private static final Schema ROW_SCHEMA =
+ Schema.builder().addInt64Field("id").addByteArrayField("data").build();
+
+ private static final Map<String, Configuration> TEST_CONFIGS =
+ ImmutableMap.of(
+ "local",
+ Configuration.of(1_000L, 5, "DirectRunner", 1_000), // approximately 1 MB
+ "small",
+ Configuration.of(1_000_000L, 20, "DataflowRunner", 1_000), // approximately 1 GB
+ "medium",
+ Configuration.of(10_000_000L, 40, "DataflowRunner", 1_000), // approximately 10 GB
+ "large",
+ Configuration.of(100_000_000L, 100, "DataflowRunner", 1_000) // approximately 100 GB
+ );
+
+ private static GcsResourceManager resourceManager;
+
+ private Configuration configuration;
+ private IcebergCatalogConfig catalogConfig;
+ private TableIdentifier tableIdentifier;
+
+ @Rule public TestPipeline writePipeline = TestPipeline.create();
+
+ @Rule public TestPipeline readPipeline = TestPipeline.create();
+
+ @BeforeClass
+ public static void beforeClass() {
+ resourceManager =
+ GcsResourceManager.builder(TestProperties.artifactBucket(), "icebergiolt", CREDENTIALS)
+ .build();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ ResourceManagerUtils.cleanResources(resourceManager);
+ }
+
+ @Before
+ public void setup() {
+ configuration = getTestConfiguration();
+
+ String uniqueSuffix =
+ DateTimeFormatter.ofPattern("MMddHHmmssSSS")
+ .withZone(ZoneOffset.UTC)
+ .format(java.time.Instant.now())
+ + "-"
+ + UUID.randomUUID().toString().substring(0, 10);
+
+ String warehouseDirectory = "icebergiolt-" + uniqueSuffix;
+ resourceManager.registerTempDir(warehouseDirectory);
+
+ String warehouseLocation =
+ String.format("gs://%s/%s", TestProperties.artifactBucket(), warehouseDirectory);
+
+ tableIdentifier = TableIdentifier.of(NAMESPACE, "table_" + uniqueSuffix.replace("-", "_"));
+
+ Map<String, String> catalogProperties =
+ ImmutableMap.of("type", "hadoop", "warehouse", warehouseLocation);
+
+ catalogConfig =
+ IcebergCatalogConfig.builder()
+ .setCatalogName("iceberg-load-test")
+ .setCatalogProperties(catalogProperties)
+ .build();
+
+ setTempLocation(writePipeline);
+ setTempLocation(readPipeline);
+ }
+
+ /** Writes generated rows to Iceberg and verifies that all rows can be read back. */
+ @Test
+ public void testIcebergWriteAndRead() throws IOException {
+ PipelineLauncher.LaunchInfo writeInfo = testWrite();
+
+ PipelineOperator.Result writeResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(writeInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(writeResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, writeInfo.jobId()));
+
+ PipelineLauncher.LaunchInfo readInfo = testRead();
+
+ PipelineOperator.Result readResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(readInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(readResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, readInfo.jobId()));
+
+ double numRecords =
+ pipelineLauncher.getMetric(
+ project,
+ region,
+ readInfo.jobId(),
+ getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME));
+
+ assertEquals((double) configuration.getNumRows(), numRecords, 0.5);
+
+ exportMetrics(writeInfo, readInfo);
+ }
+
+ private PipelineLauncher.LaunchInfo testWrite() throws IOException {
+ writePipeline
+ .apply("Generate records", GenerateSequence.from(0).to(configuration.getNumRows()))
+ .apply("Map records", ParDo.of(new MapToIcebergFormat(configuration.getValueSizeBytes())))
+ .setRowSchema(ROW_SCHEMA)
+ .apply("Write to Iceberg", IcebergIO.writeRows(catalogConfig).to(tableIdentifier));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("write-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(writePipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private PipelineLauncher.LaunchInfo testRead() throws IOException {
+ readPipeline
+ .apply("Read from Iceberg", IcebergIO.readRows(catalogConfig).from(tableIdentifier))
+ .apply("Counting element", ParDo.of(new CountingFn<>(READ_ELEMENT_METRIC_NAME)));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("read-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(readPipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private void exportMetrics(
+ PipelineLauncher.LaunchInfo writeInfo, PipelineLauncher.LaunchInfo readInfo) {
+
+ /*
+ * Only use PCollection names belonging to transforms defined directly in this test.
+ * Internal IcebergIO transform names may change when the implementation changes.
+ */
+ MetricsConfiguration writeMetricsConfig =
+ MetricsConfiguration.builder()
+ .setInputPCollection("Map records.out0")
+ .setInputPCollectionV2("Map records/ParMultiDo(MapToIcebergFormat).out0")
+ .build();
+
+ MetricsConfiguration readMetricsConfig =
+ MetricsConfiguration.builder()
+ .setOutputPCollection("Counting element.out0")
+ .setOutputPCollectionV2("Counting element/ParMultiDo(Counting).out0")
+ .build();
+
+ try {
+ exportMetricsToBigQuery(writeInfo, getMetrics(writeInfo, writeMetricsConfig));
+
+ exportMetricsToBigQuery(readInfo, getMetrics(readInfo, readMetricsConfig));
+ } catch (IOException | ParseException | InterruptedException e) {
+ throw new RuntimeException("Unable to collect or export IcebergIO load-test metrics.", e);
+ }
+ }
+
+ private Configuration getTestConfiguration() {
+ String testConfig =
+ TestProperties.getProperty("configuration", "local", TestProperties.Type.PROPERTY);
+
+ Configuration selectedConfiguration = TEST_CONFIGS.get(testConfig);
+
+ if (selectedConfiguration == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown test configuration: [%s]. Known configurations: %s",
+ testConfig, TEST_CONFIGS.keySet()));
+ }
+
+ return selectedConfiguration;
+ }
+
+ private static void setTempLocation(TestPipeline pipeline) {
+ String tempBucketName = TestProperties.artifactBucket();
+
+ if (Strings.isNullOrEmpty(tempBucketName)) {
+ return;
+ }
+
+ String tempLocation = String.format("gs://%s/temp/", tempBucketName);
+
+ pipeline.getOptions().as(TestPipelineOptions.class).setTempRoot(tempLocation);
+
+ pipeline.getOptions().setTempLocation(tempLocation);
+ }
+
+ /** Maps a generated sequence value to a deterministic Iceberg-compatible Beam row. */
+ private static class MapToIcebergFormat extends DoFn<Long, Row> {
+
+ private final int valueSizeBytes;
+
+ private MapToIcebergFormat(int valueSizeBytes) {
+ if (valueSizeBytes <= 0) {
+ throw new IllegalArgumentException("valueSizeBytes must be positive.");
+ }
+
+ this.valueSizeBytes = valueSizeBytes;
+ }
+
+ @ProcessElement
+ public void processElement(@Element Long index, OutputReceiver<Row> output) {
+
+ byte[] value = new byte[valueSizeBytes];
+
+ /*
+ * Using the row index as the seed makes retries deterministic. Random data also prevents
+ * compression from making the generated payload significantly smaller than configured.
+ */
+ new Random(index).nextBytes(value);
+
+ output.output(Row.withSchema(ROW_SCHEMA).addValues(index, value).build());
+ }
+ }
Review Comment:
Instantiating a `new Random(index)` object inside `processElement` for every single element processed by the pipeline introduces significant CPU overhead and garbage collection pressure. This is especially problematic in load and performance tests where millions or billions of elements are processed, as it can skew the benchmark results.
Instead, declare a single `transient Random` field in the `DoFn` class, lazily initialize it, and reuse it by calling `setSeed(index)` for each element to maintain determinism.
```java
private static class MapToIcebergFormat extends DoFn<Long, Row> {
private final int valueSizeBytes;
private transient Random random;
private MapToIcebergFormat(int valueSizeBytes) {
if (valueSizeBytes <= 0) {
throw new IllegalArgumentException("valueSizeBytes must be positive.");
}
this.valueSizeBytes = valueSizeBytes;
}
@ProcessElement
public void processElement(@Element Long index, OutputReceiver<Row> output) {
if (random == null) {
random = new Random();
}
byte[] value = new byte[valueSizeBytes];
/*
* Using the row index as the seed makes retries deterministic. Random data also prevents
* compression from making the generated payload significantly smaller than configured.
*/
random.setSeed(index);
random.nextBytes(value);
output.output(Row.withSchema(ROW_SCHEMA).addValues(index, value).build());
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39270:
URL: https://github.com/apache/beam/pull/39270#issuecomment-4938617015
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3616475152
##########
it/clickhouse/build.gradle:
##########
@@ -29,6 +29,12 @@ ext.summary = "Integration test utilities for ClickHouse."
def clickhouse_java_client_version = "0.9.6"
+configurations.configureEach {
+ resolutionStrategy.capabilitiesResolution.withCapability("org.lz4:lz4-java") {
+ select("at.yawk.lz4:lz4-java:1.10.2")
+ }
+}
Review Comment:
Yes, because of moving some dependencies to common, there was a lz4 conflict
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39270:
URL: https://github.com/apache/beam/pull/39270#issuecomment-4924341123
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces comprehensive load testing and benchmarking for IcebergIO. It includes a significant refactoring of existing integration test utilities to promote code reuse by moving common components into a shared module. Additionally, it adds support for IcebergIO performance tracking on the project's website, providing visibility into read and write performance metrics.
### Highlights
* **Refactoring**: Moved common integration test utilities from the gcp module to a shared common module to improve code reuse.
* **Load Testing**: Implemented new load tests for IcebergIO Batch Read and Write operations.
* **Documentation and Metrics**: Added IcebergIO performance metrics to the project website, including new Looker diagrams for visualization.
<details>
<summary><b>New Features</b></summary>
🧠You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Ignored Files</b></summary>
* **Ignored by pattern:** `.github/workflows/**` (1)
* .github/workflows/beam_PostCommit_Java_IO_Performance_Tests.yml
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3616307774
##########
it/clickhouse/build.gradle:
##########
@@ -29,6 +29,12 @@ ext.summary = "Integration test utilities for ClickHouse."
def clickhouse_java_client_version = "0.9.6"
+configurations.configureEach {
+ resolutionStrategy.capabilitiesResolution.withCapability("org.lz4:lz4-java") {
+ select("at.yawk.lz4:lz4-java:1.10.2")
+ }
+}
Review Comment:
Is this change intentional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3578723450
##########
it/common/src/main/resources/test-artifact.json:
##########
@@ -0,0 +1 @@
+["This is a test artifact."]
Review Comment:
It is used in `GcsResourceManagerTest`. I removed the other two that were not used. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on PR #39270:
URL: https://github.com/apache/beam/pull/39270#issuecomment-5024975318
@damccorm could you please re-review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto merged PR #39270:
URL: https://github.com/apache/beam/pull/39270
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39270:
URL: https://github.com/apache/beam/pull/39270#issuecomment-5025619133
Assigning reviewers:
R: @liferoad for label build.
R: @kennknowles for label website.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3561213904
##########
it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManager.java:
##########
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.bigquery;
+package org.apache.beam.it.common.bigquery;
Review Comment:
Why did we need to move all of these to common? Some of these do seem GCP specific
##########
it/common/src/main/resources/test-artifact.json:
##########
@@ -0,0 +1 @@
+["This is a test artifact."]
Review Comment:
How is this (and the similar file below) used?
##########
it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java:
##########
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.it.iceberg;
+
+import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult;
+import static org.junit.Assert.assertEquals;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.text.ParseException;
+import java.time.Duration;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import org.apache.beam.it.common.PipelineLauncher;
+import org.apache.beam.it.common.PipelineOperator;
+import org.apache.beam.it.common.TestProperties;
+import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
+import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.storage.GcsResourceManager;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.io.iceberg.IcebergIO;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestPipelineOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * IcebergIO performance tests using a Hadoop catalog backed by GCS.
+ *
+ * <p>The test writes generated rows to an Iceberg table, reads the table back, validates the number
+ * of rows, and exports write and read performance metrics.
+ *
+ * <p>Example trigger command:
+ *
+ * <pre>
+ * ./gradlew :it:iceberg:integrationTest \
+ * --tests "org.apache.beam.it.iceberg.IcebergIOLT" \
+ * -Dconfiguration=local \
+ * -Dproject=[gcpProject] \
+ * -DartifactBucket=[artifactBucket]
+ * </pre>
+ */
+public final class IcebergIOLT extends IOLoadTestBase {
+
+ private static final String READ_ELEMENT_METRIC_NAME = "read_count";
+ private static final String NAMESPACE = "default";
+
+ private static final Schema ROW_SCHEMA =
+ Schema.builder().addInt64Field("id").addByteArrayField("data").build();
+
+ private static final Map<String, Configuration> TEST_CONFIGS =
+ ImmutableMap.of(
+ "local",
+ Configuration.of(1_000L, 5, "DirectRunner", 1_000), // approximately 1 MB
+ "small",
+ Configuration.of(1_000_000L, 20, "DataflowRunner", 1_000), // approximately 1 GB
+ "medium",
+ Configuration.of(10_000_000L, 40, "DataflowRunner", 1_000), // approximately 10 GB
+ "large",
+ Configuration.of(100_000_000L, 100, "DataflowRunner", 1_000) // approximately 100 GB
+ );
+
+ private static GcsResourceManager resourceManager;
+
+ private Configuration configuration;
+ private IcebergCatalogConfig catalogConfig;
+ private TableIdentifier tableIdentifier;
+
+ @Rule public TestPipeline writePipeline = TestPipeline.create();
+
+ @Rule public TestPipeline readPipeline = TestPipeline.create();
+
+ @BeforeClass
+ public static void beforeClass() {
+ resourceManager =
+ GcsResourceManager.builder(TestProperties.artifactBucket(), "icebergiolt", CREDENTIALS)
+ .build();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ ResourceManagerUtils.cleanResources(resourceManager);
+ }
+
+ @Before
+ public void setup() {
+ configuration = getTestConfiguration();
+
+ String uniqueSuffix =
+ DateTimeFormatter.ofPattern("MMddHHmmssSSS")
+ .withZone(ZoneOffset.UTC)
+ .format(java.time.Instant.now())
+ + "-"
+ + UUID.randomUUID().toString().substring(0, 10);
+
+ String warehouseDirectory = "icebergiolt-" + uniqueSuffix;
+ resourceManager.registerTempDir(warehouseDirectory);
+
+ String warehouseLocation =
+ String.format("gs://%s/%s", TestProperties.artifactBucket(), warehouseDirectory);
+
+ tableIdentifier = TableIdentifier.of(NAMESPACE, "table_" + uniqueSuffix.replace("-", "_"));
+
+ Map<String, String> catalogProperties =
+ ImmutableMap.of("type", "hadoop", "warehouse", warehouseLocation);
+
+ catalogConfig =
+ IcebergCatalogConfig.builder()
+ .setCatalogName("iceberg-load-test")
+ .setCatalogProperties(catalogProperties)
+ .build();
+
+ setTempLocation(writePipeline);
+ setTempLocation(readPipeline);
+ }
+
+ /** Writes generated rows to Iceberg and verifies that all rows can be read back. */
+ @Test
+ public void testIcebergWriteAndRead() throws IOException {
+ PipelineLauncher.LaunchInfo writeInfo = testWrite();
+
+ PipelineOperator.Result writeResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(writeInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(writeResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, writeInfo.jobId()));
+
+ PipelineLauncher.LaunchInfo readInfo = testRead();
+
+ PipelineOperator.Result readResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(readInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(readResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, readInfo.jobId()));
+
+ double numRecords =
+ pipelineLauncher.getMetric(
+ project,
+ region,
+ readInfo.jobId(),
+ getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME));
+
+ assertEquals((double) configuration.getNumRows(), numRecords, 0.5);
+
+ exportMetrics(writeInfo, readInfo);
+ }
+
+ private PipelineLauncher.LaunchInfo testWrite() throws IOException {
+ writePipeline
+ .apply("Generate records", GenerateSequence.from(0).to(configuration.getNumRows()))
+ .apply("Map records", ParDo.of(new MapToIcebergFormat(configuration.getValueSizeBytes())))
+ .setRowSchema(ROW_SCHEMA)
+ .apply("Write to Iceberg", IcebergIO.writeRows(catalogConfig).to(tableIdentifier));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("write-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(writePipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private PipelineLauncher.LaunchInfo testRead() throws IOException {
+ readPipeline
+ .apply("Read from Iceberg", IcebergIO.readRows(catalogConfig).from(tableIdentifier))
+ .apply("Counting element", ParDo.of(new CountingFn<>(READ_ELEMENT_METRIC_NAME)));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("read-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(readPipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private void exportMetrics(
+ PipelineLauncher.LaunchInfo writeInfo, PipelineLauncher.LaunchInfo readInfo) {
+
+ /*
+ * Only use PCollection names belonging to transforms defined directly in this test.
+ * Internal IcebergIO transform names may change when the implementation changes.
+ */
+ MetricsConfiguration writeMetricsConfig =
+ MetricsConfiguration.builder()
+ .setInputPCollection("Map records.out0")
+ .setInputPCollectionV2("Map records/ParMultiDo(MapToIcebergFormat).out0")
+ .build();
+
+ MetricsConfiguration readMetricsConfig =
+ MetricsConfiguration.builder()
+ .setOutputPCollection("Counting element.out0")
+ .setOutputPCollectionV2("Counting element/ParMultiDo(Counting).out0")
+ .build();
+
+ try {
+ exportMetricsToBigQuery(writeInfo, getMetrics(writeInfo, writeMetricsConfig));
+
+ exportMetricsToBigQuery(readInfo, getMetrics(readInfo, readMetricsConfig));
+ } catch (IOException | ParseException | InterruptedException e) {
+ throw new RuntimeException("Unable to collect or export IcebergIO load-test metrics.", e);
+ }
+ }
+
+ private Configuration getTestConfiguration() {
+ String testConfig =
+ TestProperties.getProperty("configuration", "local", TestProperties.Type.PROPERTY);
+
+ Configuration selectedConfiguration = TEST_CONFIGS.get(testConfig);
+
+ if (selectedConfiguration == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown test configuration: [%s]. Known configurations: %s",
+ testConfig, TEST_CONFIGS.keySet()));
+ }
+
+ return selectedConfiguration;
+ }
+
+ private static void setTempLocation(TestPipeline pipeline) {
+ String tempBucketName = TestProperties.artifactBucket();
+
+ if (Strings.isNullOrEmpty(tempBucketName)) {
+ return;
+ }
+
+ String tempLocation = String.format("gs://%s/temp/", tempBucketName);
+
+ pipeline.getOptions().as(TestPipelineOptions.class).setTempRoot(tempLocation);
+
+ pipeline.getOptions().setTempLocation(tempLocation);
+ }
+
+ /** Maps a generated sequence value to a deterministic Iceberg-compatible Beam row. */
+ private static class MapToIcebergFormat extends DoFn<Long, Row> {
+
+ private final int valueSizeBytes;
+
+ private MapToIcebergFormat(int valueSizeBytes) {
+ if (valueSizeBytes <= 0) {
+ throw new IllegalArgumentException("valueSizeBytes must be positive.");
+ }
+
+ this.valueSizeBytes = valueSizeBytes;
+ }
+
+ @ProcessElement
+ public void processElement(@Element Long index, OutputReceiver<Row> output) {
+
+ byte[] value = new byte[valueSizeBytes];
+
+ /*
+ * Using the row index as the seed makes retries deterministic. Random data also prevents
+ * compression from making the generated payload significantly smaller than configured.
+ */
+ new Random(index).nextBytes(value);
+
+ output.output(Row.withSchema(ROW_SCHEMA).addValues(index, value).build());
+ }
+ }
Review Comment:
I think this is true - would you mind addressing this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3561472496
##########
it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManager.java:
##########
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.bigquery;
+package org.apache.beam.it.common.bigquery;
Review Comment:
I moved them so Kafka, Iceberg, and other non-GCP IO tests could reuse the load-test infrastructure without depending on the whole GCP test module. I agree some utilities are still GCP-specific, so keeping them in clearly named `gcp` or `dataflow` packages within `common` may be a better separation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IcerbergIO Load test and benchmarking [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #39270:
URL: https://github.com/apache/beam/pull/39270#discussion_r3578752733
##########
it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java:
##########
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.it.iceberg;
+
+import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult;
+import static org.junit.Assert.assertEquals;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.text.ParseException;
+import java.time.Duration;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import org.apache.beam.it.common.PipelineLauncher;
+import org.apache.beam.it.common.PipelineOperator;
+import org.apache.beam.it.common.TestProperties;
+import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
+import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.storage.GcsResourceManager;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.io.iceberg.IcebergIO;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestPipelineOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * IcebergIO performance tests using a Hadoop catalog backed by GCS.
+ *
+ * <p>The test writes generated rows to an Iceberg table, reads the table back, validates the number
+ * of rows, and exports write and read performance metrics.
+ *
+ * <p>Example trigger command:
+ *
+ * <pre>
+ * ./gradlew :it:iceberg:integrationTest \
+ * --tests "org.apache.beam.it.iceberg.IcebergIOLT" \
+ * -Dconfiguration=local \
+ * -Dproject=[gcpProject] \
+ * -DartifactBucket=[artifactBucket]
+ * </pre>
+ */
+public final class IcebergIOLT extends IOLoadTestBase {
+
+ private static final String READ_ELEMENT_METRIC_NAME = "read_count";
+ private static final String NAMESPACE = "default";
+
+ private static final Schema ROW_SCHEMA =
+ Schema.builder().addInt64Field("id").addByteArrayField("data").build();
+
+ private static final Map<String, Configuration> TEST_CONFIGS =
+ ImmutableMap.of(
+ "local",
+ Configuration.of(1_000L, 5, "DirectRunner", 1_000), // approximately 1 MB
+ "small",
+ Configuration.of(1_000_000L, 20, "DataflowRunner", 1_000), // approximately 1 GB
+ "medium",
+ Configuration.of(10_000_000L, 40, "DataflowRunner", 1_000), // approximately 10 GB
+ "large",
+ Configuration.of(100_000_000L, 100, "DataflowRunner", 1_000) // approximately 100 GB
+ );
+
+ private static GcsResourceManager resourceManager;
+
+ private Configuration configuration;
+ private IcebergCatalogConfig catalogConfig;
+ private TableIdentifier tableIdentifier;
+
+ @Rule public TestPipeline writePipeline = TestPipeline.create();
+
+ @Rule public TestPipeline readPipeline = TestPipeline.create();
+
+ @BeforeClass
+ public static void beforeClass() {
+ resourceManager =
+ GcsResourceManager.builder(TestProperties.artifactBucket(), "icebergiolt", CREDENTIALS)
+ .build();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ ResourceManagerUtils.cleanResources(resourceManager);
+ }
+
+ @Before
+ public void setup() {
+ configuration = getTestConfiguration();
+
+ String uniqueSuffix =
+ DateTimeFormatter.ofPattern("MMddHHmmssSSS")
+ .withZone(ZoneOffset.UTC)
+ .format(java.time.Instant.now())
+ + "-"
+ + UUID.randomUUID().toString().substring(0, 10);
+
+ String warehouseDirectory = "icebergiolt-" + uniqueSuffix;
+ resourceManager.registerTempDir(warehouseDirectory);
+
+ String warehouseLocation =
+ String.format("gs://%s/%s", TestProperties.artifactBucket(), warehouseDirectory);
+
+ tableIdentifier = TableIdentifier.of(NAMESPACE, "table_" + uniqueSuffix.replace("-", "_"));
+
+ Map<String, String> catalogProperties =
+ ImmutableMap.of("type", "hadoop", "warehouse", warehouseLocation);
+
+ catalogConfig =
+ IcebergCatalogConfig.builder()
+ .setCatalogName("iceberg-load-test")
+ .setCatalogProperties(catalogProperties)
+ .build();
+
+ setTempLocation(writePipeline);
+ setTempLocation(readPipeline);
+ }
+
+ /** Writes generated rows to Iceberg and verifies that all rows can be read back. */
+ @Test
+ public void testIcebergWriteAndRead() throws IOException {
+ PipelineLauncher.LaunchInfo writeInfo = testWrite();
+
+ PipelineOperator.Result writeResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(writeInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(writeResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, writeInfo.jobId()));
+
+ PipelineLauncher.LaunchInfo readInfo = testRead();
+
+ PipelineOperator.Result readResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(readInfo, Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(readResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, readInfo.jobId()));
+
+ double numRecords =
+ pipelineLauncher.getMetric(
+ project,
+ region,
+ readInfo.jobId(),
+ getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME));
+
+ assertEquals((double) configuration.getNumRows(), numRecords, 0.5);
+
+ exportMetrics(writeInfo, readInfo);
+ }
+
+ private PipelineLauncher.LaunchInfo testWrite() throws IOException {
+ writePipeline
+ .apply("Generate records", GenerateSequence.from(0).to(configuration.getNumRows()))
+ .apply("Map records", ParDo.of(new MapToIcebergFormat(configuration.getValueSizeBytes())))
+ .setRowSchema(ROW_SCHEMA)
+ .apply("Write to Iceberg", IcebergIO.writeRows(catalogConfig).to(tableIdentifier));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("write-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(writePipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private PipelineLauncher.LaunchInfo testRead() throws IOException {
+ readPipeline
+ .apply("Read from Iceberg", IcebergIO.readRows(catalogConfig).from(tableIdentifier))
+ .apply("Counting element", ParDo.of(new CountingFn<>(READ_ELEMENT_METRIC_NAME)));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("read-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(readPipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10", TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private void exportMetrics(
+ PipelineLauncher.LaunchInfo writeInfo, PipelineLauncher.LaunchInfo readInfo) {
+
+ /*
+ * Only use PCollection names belonging to transforms defined directly in this test.
+ * Internal IcebergIO transform names may change when the implementation changes.
+ */
+ MetricsConfiguration writeMetricsConfig =
+ MetricsConfiguration.builder()
+ .setInputPCollection("Map records.out0")
+ .setInputPCollectionV2("Map records/ParMultiDo(MapToIcebergFormat).out0")
+ .build();
+
+ MetricsConfiguration readMetricsConfig =
+ MetricsConfiguration.builder()
+ .setOutputPCollection("Counting element.out0")
+ .setOutputPCollectionV2("Counting element/ParMultiDo(Counting).out0")
+ .build();
+
+ try {
+ exportMetricsToBigQuery(writeInfo, getMetrics(writeInfo, writeMetricsConfig));
+
+ exportMetricsToBigQuery(readInfo, getMetrics(readInfo, readMetricsConfig));
+ } catch (IOException | ParseException | InterruptedException e) {
+ throw new RuntimeException("Unable to collect or export IcebergIO load-test metrics.", e);
+ }
+ }
+
+ private Configuration getTestConfiguration() {
+ String testConfig =
+ TestProperties.getProperty("configuration", "local", TestProperties.Type.PROPERTY);
+
+ Configuration selectedConfiguration = TEST_CONFIGS.get(testConfig);
+
+ if (selectedConfiguration == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown test configuration: [%s]. Known configurations: %s",
+ testConfig, TEST_CONFIGS.keySet()));
+ }
+
+ return selectedConfiguration;
+ }
+
+ private static void setTempLocation(TestPipeline pipeline) {
+ String tempBucketName = TestProperties.artifactBucket();
+
+ if (Strings.isNullOrEmpty(tempBucketName)) {
+ return;
+ }
+
+ String tempLocation = String.format("gs://%s/temp/", tempBucketName);
+
+ pipeline.getOptions().as(TestPipelineOptions.class).setTempRoot(tempLocation);
+
+ pipeline.getOptions().setTempLocation(tempLocation);
+ }
+
+ /** Maps a generated sequence value to a deterministic Iceberg-compatible Beam row. */
+ private static class MapToIcebergFormat extends DoFn<Long, Row> {
+
+ private final int valueSizeBytes;
+
+ private MapToIcebergFormat(int valueSizeBytes) {
+ if (valueSizeBytes <= 0) {
+ throw new IllegalArgumentException("valueSizeBytes must be positive.");
+ }
+
+ this.valueSizeBytes = valueSizeBytes;
+ }
+
+ @ProcessElement
+ public void processElement(@Element Long index, OutputReceiver<Row> output) {
+
+ byte[] value = new byte[valueSizeBytes];
+
+ /*
+ * Using the row index as the seed makes retries deterministic. Random data also prevents
+ * compression from making the generated payload significantly smaller than configured.
+ */
+ new Random(index).nextBytes(value);
+
+ output.output(Row.withSchema(ROW_SCHEMA).addValues(index, value).build());
+ }
+ }
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org