Skip to content

Commit 098cec7

Browse files
chamikaramjchamikaramj
committed
Adds a Dataflow integration test suite for Delta Lake
1 parent 6ffee9f commit 098cec7

4 files changed

Lines changed: 294 additions & 0 deletions

File tree

‎.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
{
2+
"comment": "Modify this file in a trivial way to cause this test suite to run.",
3+
"modification": 1
4+
}

‎.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one or more
2+
# contributor license agreements. See the NOTICE file distributed with
3+
# this work for additional information regarding copyright ownership.
4+
# The ASF licenses this file to You under the Apache License, Version 2.0
5+
# (the "License"); you may not use this file except in compliance with
6+
# the License. You may obtain a copy of the License at
7+
#
8+
# http://www.apache.org/licenses/LICENSE-2.0
9+
#
10+
# Unless required by applicable law or agreed to in writing, software
11+
# distributed under the License is distributed on an "AS IS" BASIS,
12+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
# See the License for the specific language governing permissions and
14+
# limitations under the License.
15+
16+
name: PostCommit Java Delta IO Dataflow
17+
18+
on:
19+
schedule:
20+
- cron: '30 4/6 * * *'
21+
pull_request_target:
22+
paths: [ 'release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json' ]
23+
workflow_dispatch:
24+
25+
# Setting explicit permissions for the action to avoid the default permissions which are `write-all` in case of pull_request_target event
26+
permissions:
27+
actions: write
28+
pull-requests: write
29+
checks: write
30+
contents: read
31+
deployments: read
32+
id-token: none
33+
issues: write
34+
discussions: read
35+
packages: read
36+
pages: read
37+
repository-projects: read
38+
security-events: read
39+
statuses: read
40+
41+
# This allows a subsequently queued workflow run to interrupt previous runs
42+
concurrency:
43+
group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.comment.id || github.event.sender.login }}'
44+
cancel-in-progress: true
45+
46+
env:
47+
DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
48+
GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
49+
GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
50+
51+
jobs:
52+
beam_PostCommit_Java_Delta_IO_Dataflow:
53+
if: |
54+
github.event_name == 'pull_request_target' ||
55+
github.event_name == 'workflow_dispatch' ||
56+
(github.event_name == 'schedule' && github.repository == 'apache/beam') ||
57+
github.event.comment.body == 'Run PostCommit Java Delta IO Dataflow'
58+
runs-on: [self-hosted, ubuntu-24.04, main]
59+
timeout-minutes: 120
60+
name: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
61+
strategy:
62+
matrix:
63+
job_name: ["beam_PostCommit_Java_Delta_IO_Dataflow"]
64+
job_phrase: ["Run PostCommit Java Delta IO Dataflow"]
65+
steps:
66+
- uses: actions/checkout@v6
67+
- name: Setup repository
68+
uses: ./.github/actions/setup-action
69+
with:
70+
comment_phrase: ${{ matrix.job_phrase }}
71+
github_token: ${{ secrets.GITHUB_TOKEN }}
72+
github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
73+
- name: Setup environment
74+
uses: ./.github/actions/setup-environment-action
75+
- name: Run PostCommit Java Delta IO Dataflow
76+
uses: ./.github/actions/gradle-command-self-hosted-action
77+
with:
78+
gradle-command: :sdks:java:io:delta:dataflowIntegrationTest --info

‎sdks/java/io/delta/build.gradle‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
* limitations under the License.
1717
*/
1818

19+
import groovy.json.JsonOutput
20+
1921
plugins { id 'org.apache.beam.module' }
2022
applyJavaNature(
2123
automaticModuleName: 'org.apache.beam.sdk.io.delta',
@@ -68,6 +70,10 @@ dependencies {
6870
testImplementation project(":sdks:java:managed")
6971
testRuntimeOnly "org.yaml:snakeyaml:2.0"
7072
testImplementation project(path: ":runners:direct-java", configuration: "shadow")
73+
testImplementation project(":sdks:java:io:google-cloud-platform")
74+
testImplementation project(":sdks:java:extensions:google-cloud-platform-core")
75+
testImplementation library.java.google_cloud_storage
76+
testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java")
7177
}
7278

7379
configurations.all {
@@ -82,3 +88,35 @@ configurations.all {
8288
resolutionStrategy.force "com.google.cloud.bigdataoss:gcsio:$bigdataoss_gcs_connector_version"
8389
resolutionStrategy.force "com.google.cloud.bigdataoss:util:$bigdataoss_gcs_connector_version"
8490
}
91+
92+
def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
93+
def gcpTempLocation = project.findProperty('gcpTempLocation') ?: 'gs://apache-beam-testing-delta-lake/temp'
94+
95+
// Used by Dataflow integration tests.
96+
task dataflowIntegrationTest(type: Test) {
97+
group = "Verification"
98+
evaluationDependsOn(":runners:google-cloud-dataflow-java")
99+
dependsOn ":runners:google-cloud-dataflow-java:buildAndPushDockerJavaContainer"
100+
finalizedBy ":runners:google-cloud-dataflow-java:cleanUpDockerJavaImages"
101+
def dockerJavaImageName = project.project(':runners:google-cloud-dataflow-java').ext.dockerJavaImageName
102+
103+
def args = [
104+
"--runner=DataflowRunner",
105+
"--region=us-central1",
106+
"--project=${gcpProject}",
107+
"--tempLocation=${gcpTempLocation}",
108+
"--tempRoot=${gcpTempLocation}",
109+
"--sdkContainerImage=${dockerJavaImageName}",
110+
"--experiments=use_runner_v2,use_staged_dataflow_worker_jar"
111+
]
112+
systemProperty "beamTestPipelineOptions", JsonOutput.toJson(args)
113+
114+
// Disable Gradle cache: these ITs interact with live service that should always be considered "out of date"
115+
outputs.upToDateWhen { false }
116+
117+
include '**/*IT.class'
118+
119+
maxParallelForks 4
120+
classpath = sourceSets.test.runtimeClasspath
121+
testClassesDirs = sourceSets.test.output.classesDirs
122+
}

‎sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.sdk.io.delta;
19+
20+
import static org.junit.Assert.assertNotNull;
21+
22+
import com.google.cloud.storage.Blob;
23+
import com.google.cloud.storage.BlobId;
24+
import com.google.cloud.storage.BlobInfo;
25+
import com.google.cloud.storage.Storage;
26+
import com.google.cloud.storage.StorageOptions;
27+
import java.nio.charset.StandardCharsets;
28+
import java.util.HashMap;
29+
import java.util.List;
30+
import java.util.Map;
31+
import java.util.stream.Collectors;
32+
import java.util.stream.IntStream;
33+
import org.apache.avro.generic.GenericRecord;
34+
import org.apache.beam.runners.direct.DirectOptions;
35+
import org.apache.beam.runners.direct.DirectRunner;
36+
import org.apache.beam.sdk.Pipeline;
37+
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
38+
import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
39+
import org.apache.beam.sdk.io.Compression;
40+
import org.apache.beam.sdk.io.FileIO;
41+
import org.apache.beam.sdk.io.parquet.ParquetIO;
42+
import org.apache.beam.sdk.managed.Managed;
43+
import org.apache.beam.sdk.options.PipelineOptionsFactory;
44+
import org.apache.beam.sdk.schemas.Schema;
45+
import org.apache.beam.sdk.testing.PAssert;
46+
import org.apache.beam.sdk.testing.TestPipeline;
47+
import org.apache.beam.sdk.transforms.Create;
48+
import org.apache.beam.sdk.transforms.MapElements;
49+
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
50+
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
51+
import org.apache.beam.sdk.values.PCollection;
52+
import org.apache.beam.sdk.values.Row;
53+
import org.apache.beam.sdk.values.TypeDescriptor;
54+
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
55+
import org.junit.After;
56+
import org.junit.Before;
57+
import org.junit.Rule;
58+
import org.junit.Test;
59+
import org.junit.rules.TestName;
60+
import org.junit.runner.RunWith;
61+
import org.junit.runners.JUnit4;
62+
import org.slf4j.Logger;
63+
import org.slf4j.LoggerFactory;
64+
65+
/** Integration tests for {@link DeltaIO}. */
66+
@RunWith(JUnit4.class)
67+
public class DeltaIOIT {
68+
private static final Logger LOG = LoggerFactory.getLogger(DeltaIOIT.class);
69+
70+
private static final String BUCKET = "apache-beam-testing-delta-lake";
71+
72+
@Rule public final TestPipeline readPipeline = TestPipeline.create();
73+
@Rule public final TestName testName = new TestName();
74+
75+
private String repoPath;
76+
private String repoPrefix;
77+
private Storage storage;
78+
79+
private static final Schema ROW_SCHEMA =
80+
Schema.builder().addInt32Field("id").addStringField("name").build();
81+
82+
private static final List<Row> TEST_ROWS =
83+
IntStream.range(0, 100)
84+
.mapToObj(i -> Row.withSchema(ROW_SCHEMA).addValues(i, "name_" + i).build())
85+
.collect(Collectors.toList());
86+
87+
@Before
88+
public void setup() throws Exception {
89+
storage = StorageOptions.newBuilder().build().getService();
90+
long salt = System.currentTimeMillis();
91+
repoPrefix = "delta_io_it/" + testName.getMethodName() + "-" + salt;
92+
repoPath = "gs://" + BUCKET + "/" + repoPrefix;
93+
94+
LOG.info("Generating Delta Lake repository at {}", repoPath);
95+
96+
// 1. Write Parquet file using a direct local pipeline
97+
DirectOptions setupOptions = PipelineOptionsFactory.as(DirectOptions.class);
98+
setupOptions.setRunner(DirectRunner.class);
99+
setupOptions.setBlockOnRun(true);
100+
Pipeline setupPipeline = Pipeline.create(setupOptions);
101+
102+
org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA);
103+
setupPipeline
104+
.apply(Create.of(TEST_ROWS).withRowSchema(ROW_SCHEMA))
105+
.apply(
106+
MapElements.into(TypeDescriptor.of(GenericRecord.class))
107+
.via(AvroUtils.getRowToGenericRecordFunction(avroSchema)))
108+
.setCoder(AvroCoder.of(avroSchema))
109+
.apply(
110+
FileIO.<GenericRecord>write()
111+
.via(ParquetIO.sink(avroSchema))
112+
.to(repoPath + "/")
113+
.withNaming(
114+
(BoundedWindow window,
115+
PaneInfo paneInfo,
116+
int numShards,
117+
int shardIndex,
118+
Compression compression) -> "part-00000.parquet"));
119+
setupPipeline.run().waitUntilFinish();
120+
121+
// 2. Find written Parquet file to inspect its size
122+
BlobId parquetBlobId = BlobId.of(BUCKET, repoPrefix + "/part-00000.parquet");
123+
Blob parquetBlob = storage.get(parquetBlobId);
124+
assertNotNull("Parquet file not found on GCS: " + parquetBlobId, parquetBlob);
125+
long fileSize = parquetBlob.getSize();
126+
127+
// 3. Create the Delta log commit file
128+
String commitContent =
129+
"{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n"
130+
+ "{\"metaData\":{\"id\":\""
131+
+ salt
132+
+ "\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n"
133+
+ "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":"
134+
+ fileSize
135+
+ ",\"modificationTime\":123456789,\"dataChange\":true}}";
136+
137+
BlobId commitBlobId = BlobId.of(BUCKET, repoPrefix + "/_delta_log/00000000000000000000.json");
138+
BlobInfo commitBlobInfo =
139+
BlobInfo.newBuilder(commitBlobId).setContentType("application/json").build();
140+
storage.create(commitBlobInfo, commitContent.getBytes(StandardCharsets.UTF_8));
141+
LOG.info("Successfully generated Delta Lake repository");
142+
}
143+
144+
@After
145+
public void teardown() {
146+
LOG.info("Cleaning up Delta Lake repository at {}", repoPath);
147+
try {
148+
Iterable<Blob> blobs =
149+
storage.list(BUCKET, Storage.BlobListOption.prefix(repoPrefix)).getValues();
150+
blobs.forEach(b -> storage.delete(b.getBlobId()));
151+
storage.close();
152+
} catch (Exception e) {
153+
LOG.warn("Failed to clean up GCS repository at {}", repoPath, e);
154+
}
155+
}
156+
157+
@Test
158+
public void testReadDeltaLakeTable() {
159+
Map<String, String> hadoopConfig = new HashMap<>();
160+
hadoopConfig.put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem");
161+
hadoopConfig.put(
162+
"fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS");
163+
164+
PCollection<Row> output =
165+
readPipeline
166+
.apply(
167+
Managed.read(Managed.DELTA_LAKE)
168+
.withConfig(ImmutableMap.of("table", repoPath, "hadoop_config", hadoopConfig)))
169+
.getSinglePCollection();
170+
171+
PAssert.that(output).containsInAnyOrder(TEST_ROWS);
172+
readPipeline.run().waitUntilFinish();
173+
}
174+
}

0 commit comments

Comments
 (0)