Skip to content

Commit 972074e

Browse files
CoderYellowCoderYellow
committed
feat: enhance Spark example with HMS and Parquet table support
- Added Iceberg HMS-backed table creation and verification in Spark example. - Introduced Hive metastore-backed Parquet table creation stored on S3, with metadata verification. - Updated `log4j2.xml` for improved logging configuration. - Extended JVM arguments for enhanced compatibility in Spark tests. - Enhanced testing utilities with additional assertions for Hive Metastore and Parquet functionality.
1 parent 8153ba5 commit 972074e

4 files changed

Lines changed: 107 additions & 12 deletions

File tree

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,7 @@ GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spring:bootRun --args='--sprin
203203

204204
The JUnit examples in `example/junit` are annotated with `@Disabled`; remove that annotation from an example class to start the configured stack.
205205

206-
The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, then uses `extensions` config to create the S3 JCEKS file on HDFS and produce Avro records to Kafka. HDFS is used as the config store for the JCEKS file; S3 and GCS are the object stores used by the Iceberg smoke checks.
206+
The Spark example in `example/spark` creates a `SparkSession` from `BigDataTestKit` endpoints, starts HDFS, Hive Metastore, Kafka, Schema Registry, LocalStack S3, and fake GCS, then uses `extensions` config to create the S3 JCEKS file on HDFS and produce Avro records to Kafka. HDFS is used as the config store for the JCEKS file; S3 and GCS are object-store Iceberg smoke checks through Hadoop catalogs. The example also creates an HMS-backed Iceberg table and a Hive metastore Parquet table stored on S3, then verifies the HMS database/table metadata through `HiveMetaStoreClient`.
207207

208208
```bash
209209
GRADLE_USER_HOME=/data/.gradle ./gradlew :example:spark:test

‎example/spark/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,5 +44,6 @@ tasks.withType<Test>().configureEach {
4444
jvmArgs(
4545
"--add-exports=java.base/sun.nio.ch=ALL-UNNAMED",
4646
"--add-opens=java.base/java.nio=ALL-UNNAMED",
47+
"--add-opens=java.base/java.net=ALL-UNNAMED",
4748
)
4849
}

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

Lines changed: 91 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package org.openprojectx.bigdata.test.example.spark
22

3+
import org.apache.hadoop.hive.conf.HiveConf
4+
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient
35
import org.apache.spark.sql.SparkSession
46
import org.junit.jupiter.api.Test
57
import org.openprojectx.bigdata.test.core.BigDataService
@@ -65,6 +67,19 @@ class SparkBigDataTestExample {
6567
storageName = "gcs",
6668
dataPath = "gs://$gcsBucket/data/demo_$runId/events_gcs",
6769
)
70+
assertIcebergTable(spark, catalog = "hms", namespace = "hms_demo_$runId", table = "events_hms", storageName = "hms")
71+
assertHiveMetastoreTable(
72+
hiveMetastoreUri = hiveMetastore.property("hive.metastore.uris"),
73+
database = "hms_demo_$runId",
74+
table = "events_hms",
75+
)
76+
assertHiveS3ParquetTable(
77+
spark = spark,
78+
hiveMetastoreUri = hiveMetastore.property("hive.metastore.uris"),
79+
database = "hive_s3_demo_$runId",
80+
table = "events_parquet_s3",
81+
location = "s3a://$s3Bucket/hive-parquet/events_parquet_s3",
82+
)
6883
}
6984
}
7085

@@ -81,21 +96,27 @@ class SparkBigDataTestExample {
8196
.appName("bigdata-test-spark-example")
8297
.master("local[2]")
8398
.config("spark.ui.enabled", "false")
84-
.config("spark.driver.extraJavaOptions", "--add-opens=java.base/java.nio=ALL-UNNAMED")
85-
.config("spark.executor.extraJavaOptions", "--add-opens=java.base/java.nio=ALL-UNNAMED")
99+
.config("spark.driver.extraJavaOptions", "--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED")
100+
.config("spark.executor.extraJavaOptions", "--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED")
86101
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
87102
.config("spark.sql.catalog.s3", "org.apache.iceberg.spark.SparkCatalog")
88103
.config("spark.sql.catalog.s3.type", "hadoop")
89104
.config("spark.sql.catalog.s3.warehouse", "s3a://$s3Bucket/warehouse")
90105
.config("spark.sql.catalog.gcs_local", "org.apache.iceberg.spark.SparkCatalog")
91106
.config("spark.sql.catalog.gcs_local.type", "hadoop")
92107
.config("spark.sql.catalog.gcs_local.warehouse", "file:${Files.createTempDirectory("bigdata-test-gcs-iceberg-warehouse-")}")
108+
.config("spark.sql.catalog.hms", "org.apache.iceberg.spark.SparkCatalog")
109+
.config("spark.sql.catalog.hms.type", "hive")
110+
.config("spark.sql.catalog.hms.uri", hiveMetastoreUri)
111+
.config("spark.sql.catalog.hms.warehouse", "file:${Files.createTempDirectory("bigdata-test-hms-warehouse-")}")
93112
.config("spark.sql.warehouse.dir", "file:${Files.createTempDirectory("bigdata-test-spark-warehouse-")}")
94113
.config("hive.metastore.uris", hiveMetastoreUri)
95114
.config("spark.hadoop.fs.defaultFS", hdfsUri)
96115
.config("spark.hadoop.hadoop.security.credential.provider.path", s3CredentialProviderPath)
97116
.config("spark.hadoop.fs.s3a.endpoint", s3Endpoint)
98117
.config("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider")
118+
.config("spark.hadoop.fs.s3a.access.key", "test")
119+
.config("spark.hadoop.fs.s3a.secret.key", "test")
99120
.config("spark.hadoop.fs.s3a.path.style.access", "true")
100121
.config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
101122
.config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
@@ -154,6 +175,74 @@ class SparkBigDataTestExample {
154175
check(count == 2L) { "Expected two Iceberg rows in $identifier" }
155176
}
156177

178+
private fun assertHiveMetastoreTable(hiveMetastoreUri: String, database: String, table: String) {
179+
val conf = HiveConf()
180+
conf.setVar(HiveConf.ConfVars.METASTOREURIS, hiveMetastoreUri)
181+
val client = HiveMetaStoreClient(conf)
182+
try {
183+
check(client.getAllDatabases().contains(database)) { "Expected HMS database $database" }
184+
val hmsTable = client.getTable(database, table)
185+
check(hmsTable.dbName == database) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
186+
check(hmsTable.tableName == table) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
187+
check(hmsTable.parameters["table_type"] == "ICEBERG") {
188+
"Expected HMS table $database.$table to be an Iceberg table, parameters=${hmsTable.parameters}"
189+
}
190+
check(hmsTable.parameters.containsKey("metadata_location")) {
191+
"Expected HMS table $database.$table to contain Iceberg metadata_location"
192+
}
193+
} finally {
194+
client.close()
195+
}
196+
}
197+
198+
private fun assertHiveS3ParquetTable(
199+
spark: SparkSession,
200+
hiveMetastoreUri: String,
201+
database: String,
202+
table: String,
203+
location: String,
204+
) {
205+
val identifier = "$database.$table"
206+
spark.sql("CREATE DATABASE IF NOT EXISTS $database")
207+
spark.sql(
208+
"""
209+
CREATE TABLE $identifier (
210+
id INT,
211+
name STRING,
212+
storage STRING
213+
) USING PARQUET
214+
LOCATION '$location'
215+
""".trimIndent(),
216+
)
217+
spark.sql("INSERT INTO $identifier VALUES (1, 'alpha', 'hive-s3-parquet'), (2, 'beta', 'hive-s3-parquet')")
218+
val count = spark.table(identifier).where("storage = 'hive-s3-parquet'").count()
219+
check(count == 2L) { "Expected two Hive S3 Parquet rows in $identifier" }
220+
assertS3ParquetFiles(spark, location)
221+
assertHiveMetastoreParquetS3Table(hiveMetastoreUri, database, table)
222+
}
223+
224+
private fun assertS3ParquetFiles(spark: SparkSession, location: String) {
225+
val files = org.apache.hadoop.fs.FileSystem.get(URI.create(location), spark.sparkContext().hadoopConfiguration())
226+
.use { fs -> fs.listStatus(org.apache.hadoop.fs.Path(location)).map { it.path.name } }
227+
check(files.any { it.endsWith(".parquet") }) { "Expected Parquet files under $location, got $files" }
228+
}
229+
230+
private fun assertHiveMetastoreParquetS3Table(hiveMetastoreUri: String, database: String, table: String) {
231+
val conf = HiveConf()
232+
conf.setVar(HiveConf.ConfVars.METASTOREURIS, hiveMetastoreUri)
233+
val client = HiveMetaStoreClient(conf)
234+
try {
235+
val hmsTable = client.getTable(database, table)
236+
check(hmsTable.dbName == database) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
237+
check(hmsTable.tableName == table) { "Expected HMS table $database.$table, got ${hmsTable.dbName}.${hmsTable.tableName}" }
238+
val provider = hmsTable.parameters["spark.sql.sources.provider"] ?: hmsTable.parameters["provider"]
239+
check(provider.equals("parquet", ignoreCase = true) || hmsTable.sd.inputFormat.contains("Parquet")) {
240+
"Expected HMS table $database.$table to be Parquet, parameters=${hmsTable.parameters}, inputFormat=${hmsTable.sd.inputFormat}"
241+
}
242+
} finally {
243+
client.close()
244+
}
245+
}
157246

158247
private fun createS3Bucket(endpoint: String, bucket: String) {
159248
val request = HttpRequest.newBuilder(URI.create("$endpoint/$bucket"))

‎example/spark/src/test/resources/log4j2.xml‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
11
<?xml version="1.0" encoding="UTF-8"?>
22
<Configuration status="WARN" monitorInterval="30">
33
<Properties>
4-
<Property name="CONSOLE_PATTERN">%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight{%-5level}{FATAL=red, ERROR=red, WARN=yellow, INFO=green, DEBUG=cyan, TRACE=blue} [%style{%20.20t}{magenta}] %style{%logger}{cyan} - %msg%n%throwable</Property>
4+
<Property name="CONSOLE_PATTERN">%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight{%-5level}{FATAL=red, ERROR=red,
5+
WARN=yellow, INFO=green, DEBUG=cyan, TRACE=blue} [%style{%20.20t}{magenta}] %style{%logger}{cyan} -
6+
%msg%n%throwable
7+
</Property>
58
</Properties>
69

710
<Appenders>
@@ -16,14 +19,16 @@
1619
<AppenderRef ref="Console"/>
1720
</Logger>
1821

19-
<!-- <Logger name="org.apache.spark.scheduler" level="DEBUG"/>-->
20-
<!-- <Logger name="org.apache.spark.scheduler.TaskSchedulerImpl" level="DEBUG"/>-->
21-
<!-- <Logger name="org.apache.spark.ExecutorAllocationManager" level="DEBUG"/>-->
22-
<!-- <Logger name="org.apache.spark.deploy.k8s" level="DEBUG"/>-->
23-
<!-- <Logger name="org.apache.spark.scheduler.cluster.k8s" level="DEBUG"/>-->
24-
<!-- <Logger name="org.apache.iceberg" level="DEBUG"/>-->
25-
<!-- <Logger name="org.apache.iceberg.rest" level="DEBUG"/>-->
26-
<!-- <Logger name="org.apache.hadoop.hive" level="DEBUG"/>-->
22+
<!-- <Logger name="org.apache.spark.scheduler" level="DEBUG"/>-->
23+
<!-- <Logger name="org.apache.spark.scheduler.TaskSchedulerImpl" level="DEBUG"/>-->
24+
<!-- <Logger name="org.apache.spark.ExecutorAllocationManager" level="DEBUG"/>-->
25+
<!-- <Logger name="org.apache.spark.deploy.k8s" level="DEBUG"/>-->
26+
<!-- <Logger name="org.apache.spark.scheduler.cluster.k8s" level="DEBUG"/>-->
27+
<!-- <Logger name="org.apache.iceberg" level="DEBUG"/>-->
28+
<!-- <Logger name="org.apache.iceberg.rest" level="DEBUG"/>-->
29+
<!-- <Logger name="org.apache.hadoop.hive" level="DEBUG"/>-->
30+
31+
<Logger name="org.apache.hadoop.util" level="TRACE"/>
2732

2833
<Root level="INFO">
2934
<AppenderRef ref="Console"/>

0 commit comments

Comments
 (0)