Skip to content

Commit 6a71f78

Browse files
CoderYellowCoderYellow
committed
feat(extensions): add S3 and GCS upload extensions to Spark tests
- Added `s3UploadExtension` and `gcsUploadExtension` for uploading JSON files to S3 and GCS during Spark test setup. - Updated tests to validate uploaded JSON input for both storage types. - Enhanced user documentation with TOML configuration guidance for upload sources. - Included examples showcasing explicit and implicit key handling in upload paths.
1 parent 0e6ab35 commit 6a71f78

5 files changed

Lines changed: 144 additions & 7 deletions

File tree

‎doc/user-guide.adoc‎

Lines changed: 47 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1357,22 +1357,32 @@ They use the AWS S3 SDK and Google Cloud Storage SDK against the endpoints expos
13571357
id = "seed-s3"
13581358
bucket = "demo-s3"
13591359
prefix = "input"
1360-
sources = [
1361-
{ source = "file:src/test/resources/data/users.json", key = "input/users.json", contentType = "application/json" },
1362-
{ source = "src/test/resources/data/events", prefix = "events" }
1363-
]
1360+
1361+
[[s3Uploads.sources]]
1362+
source = "file:src/test/resources/data/users.json"
1363+
key = "input/users.json"
1364+
contentType = "application/json"
1365+
1366+
[[s3Uploads.sources]]
1367+
source = "src/test/resources/data/events"
1368+
prefix = "events"
13641369
13651370
[[gcsUploads]]
13661371
id = "seed-gcs"
13671372
bucket = "demo-gcs"
13681373
project = "bigdata-test"
13691374
prefix = "input"
13701375
createBucket = true
1371-
sources = [
1372-
{ source = "classpath:data/products.json", key = "input/products.json", contentType = "application/json" }
1373-
]
1376+
1377+
[[gcsUploads.sources]]
1378+
source = "classpath:data/products.json"
1379+
key = "input/products.json"
1380+
contentType = "application/json"
13741381
----
13751382

1383+
Use `[[s3Uploads.sources]]` and `[[gcsUploads.sources]]` for upload sources.
1384+
Do not use an inline `sources = [ ... ]` array in project examples; the documented and tested TOML form is a table array, which is easier to extend and avoids parser differences around inline tables.
1385+
13761386
Programmatic config:
13771387

13781388
[source,kotlin]
@@ -1397,6 +1407,36 @@ Directory uploads should use filesystem paths because classpath directories are
13971407
When `key` is omitted for a file, the object key is built from the extension `prefix`, source `prefix`, and file name.
13981408
For directories, object keys are built from the extension `prefix`, source `prefix`, and the relative file path under the directory.
13991409

1410+
Path mapping:
1411+
1412+
[cols="1,1,1,1"]
1413+
|===
1414+
|Config |Source type |Final object key |Readable URI
1415+
1416+
|`bucket = "demo-s3"`, extension `prefix = "input"`, source `key = "input/users.json"`
1417+
|Single file with explicit `key`
1418+
|`input/users.json`
1419+
|`s3a://demo-s3/input/users.json`
1420+
1421+
|`bucket = "demo-s3"`, extension `prefix = "input"`, source `source = "file:.../users.json"`, no source `key` or source `prefix`
1422+
|Single file with implicit key
1423+
|`input/users.json`
1424+
|`s3a://demo-s3/input/users.json`
1425+
1426+
|`bucket = "demo-s3"`, extension `prefix = "input"`, source `prefix = "events"`, directory file `2026/01.json`
1427+
|Directory
1428+
|`input/events/2026/01.json`
1429+
|`s3a://demo-s3/input/events/2026/01.json`
1430+
1431+
|`bucket = "demo-gcs"`, extension `prefix = "input"`, source `key = "input/products.json"`
1432+
|Single file with explicit `key`
1433+
|`input/products.json`
1434+
|`gs://demo-gcs/input/products.json`
1435+
|===
1436+
1437+
`key` is already the full object key inside the bucket. It is not prefixed again by the extension `prefix`.
1438+
Use either `key` for an exact object path, or omit `key` and let the extension combine `prefix` values with the file name or directory-relative path.
1439+
14001440
Outputs:
14011441

14021442
* `<id>.bucket`

‎example/spark/src/test/kotlin/org/openprojectx/bigdata/test/example/spark/SparkBigDataScenario.kt‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@ abstract class SparkBigDataScenario {
1616
protected abstract val s3BucketExtensionId: String
1717
protected abstract val gcsBucketExtensionId: String
1818
protected open val sparkSqlPrepExtensionId: String = "spark-sql-prep"
19+
protected open val s3UploadExtensionId: String = "spark-s3-upload"
20+
protected open val gcsUploadExtensionId: String = "spark-gcs-upload"
21+
protected open val s3TomlUploadExtensionId: String = "spark-s3-upload-toml"
22+
protected open val gcsTomlUploadExtensionId: String = "spark-gcs-upload-toml"
1923

2024
@Test
2125
fun runsSparkScenario(kit: BigDataTestKit, extensions: BigDataExtensionResult) {
@@ -52,6 +56,34 @@ abstract class SparkBigDataScenario {
5256
sslProperties = context.environment.kafkaSslProperties,
5357
)
5458
},
59+
SparkScenarioCheck { context ->
60+
assertUploadedJsonInput(
61+
spark = context.spark,
62+
path = "s3a://${context.environment.s3Bucket}/${context.environment.s3UploadedObjectKey}",
63+
storageName = "s3-upload",
64+
)
65+
},
66+
SparkScenarioCheck { context ->
67+
assertUploadedJsonInput(
68+
spark = context.spark,
69+
path = "gs://${context.environment.gcsBucket}/${context.environment.gcsUploadedObjectKey}",
70+
storageName = "gcs-upload",
71+
)
72+
},
73+
SparkScenarioCheck { context ->
74+
assertUploadedJsonInput(
75+
spark = context.spark,
76+
path = "s3a://${context.environment.s3TomlUploadBucket}/${context.environment.s3TomlUploadedObjectKey}",
77+
storageName = "s3-upload-toml",
78+
)
79+
},
80+
SparkScenarioCheck { context ->
81+
assertUploadedJsonInput(
82+
spark = context.spark,
83+
path = "gs://${context.environment.gcsTomlUploadBucket}/${context.environment.gcsTomlUploadedObjectKey}",
84+
storageName = "gcs-upload-toml",
85+
)
86+
},
5587
)
5688

5789
protected open fun targetChecks(): List<SparkScenarioCheck> =
@@ -146,6 +178,12 @@ abstract class SparkBigDataScenario {
146178
gcsEndpoint = gcs.property("google.cloud.storage.host"),
147179
gcsBucket = gcsBucket,
148180
gcsIcebergDataPath = "${extensions.required("$gcsBucketExtensionId.gs.uri")}/data/demo_$runId/events_gcs",
181+
s3UploadedObjectKey = extensions.required("$s3UploadExtensionId.uploaded.0.key"),
182+
gcsUploadedObjectKey = extensions.required("$gcsUploadExtensionId.uploaded.0.key"),
183+
s3TomlUploadBucket = extensions.required("$s3TomlUploadExtensionId.bucket"),
184+
s3TomlUploadedObjectKey = extensions.required("$s3TomlUploadExtensionId.uploaded.0.key"),
185+
gcsTomlUploadBucket = extensions.required("$gcsTomlUploadExtensionId.bucket"),
186+
gcsTomlUploadedObjectKey = extensions.required("$gcsTomlUploadExtensionId.uploaded.0.key"),
149187
sparkSqlPrepExecutedStatements = extensions.optional("$sparkSqlPrepExtensionId.executed-statements"),
150188
s3CredentialProviderPath = extensions.required("s3-jceks.credential-provider.path"),
151189
s3CredentialProviderHdfsPath = extensions.required("s3-jceks.hdfs.path"),
@@ -314,6 +352,15 @@ abstract class SparkBigDataScenario {
314352
check(rows.count() == 2L) { "Expected two Avro Kafka records in $topic" }
315353
}
316354

355+
protected fun assertUploadedJsonInput(
356+
spark: SparkSession,
357+
path: String,
358+
storageName: String,
359+
) {
360+
val count = spark.read().json(path).where("storage = 'upload-seed'").count()
361+
check(count == 2L) { "Expected two $storageName uploaded JSON rows in $path" }
362+
}
363+
317364
protected fun assertSparkSqlPreparation(
318365
spark: SparkSession,
319366
environment: SparkScenarioEnvironment,
@@ -543,6 +590,12 @@ data class SparkScenarioEnvironment(
543590
val gcsEndpoint: String,
544591
val gcsBucket: String,
545592
val gcsIcebergDataPath: String,
593+
val s3UploadedObjectKey: String,
594+
val gcsUploadedObjectKey: String,
595+
val s3TomlUploadBucket: String,
596+
val s3TomlUploadedObjectKey: String,
597+
val gcsTomlUploadBucket: String,
598+
val gcsTomlUploadedObjectKey: String,
546599
val sparkSqlPrepExecutedStatements: String?,
547600
val s3CredentialProviderPath: String,
548601
val s3CredentialProviderHdfsPath: String,

‎example/spark/src/test/kotlin/org/openprojectx/bigdata/test/example/spark/SparkBigDataTestExample.kt‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,16 @@ class SparkBigDataTestExample : SparkBigDataScenario() {
1717
override val s3BucketExtensionId: String get() = S3_BUCKET_ID
1818
override val gcsBucketExtensionId: String get() = GCS_BUCKET_ID
1919
override val sparkSqlPrepExtensionId: String get() = SPARK_SQL_PREP_ID
20+
override val s3UploadExtensionId: String get() = S3_UPLOAD_ID
21+
override val gcsUploadExtensionId: String get() = GCS_UPLOAD_ID
2022

2123
companion object : BigDataExtensionsConfigurer {
2224
private val scenarioRunId = System.nanoTime().toString()
2325
private const val S3_BUCKET_ID = "spark-s3-bucket"
2426
private const val GCS_BUCKET_ID = "spark-gcs-bucket"
2527
private const val SPARK_SQL_PREP_ID = "spark-sql-prep"
28+
private const val S3_UPLOAD_ID = "spark-s3-upload"
29+
private const val GCS_UPLOAD_ID = "spark-gcs-upload"
2630

2731
override fun configure(extensions: BigDataExtensionsBuilder) {
2832
val s3Bucket = "spark-iceberg-s3-$scenarioRunId"
@@ -32,6 +36,24 @@ class SparkBigDataTestExample : SparkBigDataScenario() {
3236

3337
extensions.s3Bucket(bucket = s3Bucket, id = S3_BUCKET_ID)
3438
extensions.gcsBucket(bucket = gcsBucket, id = GCS_BUCKET_ID)
39+
extensions.s3Upload(s3Bucket) {
40+
id = S3_UPLOAD_ID
41+
prefix = "uploaded"
42+
file(
43+
source = "classpath:data/upload-seed.json",
44+
key = "uploaded/s3-seed.json",
45+
contentType = "application/json",
46+
)
47+
}
48+
extensions.gcsUpload(gcsBucket) {
49+
id = GCS_UPLOAD_ID
50+
prefix = "uploaded"
51+
file(
52+
source = "classpath:data/upload-seed.json",
53+
key = "uploaded/gcs-seed.json",
54+
contentType = "application/json",
55+
)
56+
}
3557
extensions.sparkSqlPreparation {
3658
id = SPARK_SQL_PREP_ID
3759
config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")

‎example/spark/src/test/resources/data/upload-seed.json‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
{"id":1,"name":"alpha","storage":"upload-seed"}
2+
{"id":2,"name":"beta","storage":"upload-seed"}

‎example/spark/src/test/resources/spark-bigdata-extensions.toml‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,3 +13,23 @@ records = [
1313
{ key = "alpha", value = { id = 1, name = "alpha" } },
1414
{ key = "beta", value = { id = 2, name = "beta" } },
1515
]
16+
17+
[[s3Uploads]]
18+
id = "spark-s3-upload-toml"
19+
bucket = "spark-upload-toml-s3"
20+
prefix = "uploaded-toml"
21+
22+
[[s3Uploads.sources]]
23+
source = "classpath:data/upload-seed.json"
24+
key = "uploaded-toml/s3-seed.json"
25+
contentType = "application/json"
26+
27+
[[gcsUploads]]
28+
id = "spark-gcs-upload-toml"
29+
bucket = "spark-upload-toml-gcs"
30+
prefix = "uploaded-toml"
31+
32+
[[gcsUploads.sources]]
33+
source = "classpath:data/upload-seed.json"
34+
key = "uploaded-toml/gcs-seed.json"
35+
contentType = "application/json"

0 commit comments

Comments
 (0)