mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-10 00:20:42 +02:00
Complete Spark integration test suite
This commit is contained in:
chrislu
@@ -0,0 +1,143 @@
|
||||
package seaweed.spark;
|
||||
|
||||
import org.apache.spark.sql.Dataset;
|
||||
import org.apache.spark.sql.Row;
|
||||
import org.apache.spark.sql.SaveMode;
|
||||
import org.apache.spark.sql.SparkSession;
|
||||
|
||||
/**
|
||||
* Example Spark application demonstrating SeaweedFS integration.
|
||||
*
|
||||
* This can be submitted to a Spark cluster using spark-submit.
|
||||
*
|
||||
* Example usage:
|
||||
* spark-submit \
|
||||
* --class seaweed.spark.SparkSeaweedFSExample \
|
||||
* --master local[2] \
|
||||
* --conf spark.hadoop.fs.seaweedfs.impl=seaweed.hdfs.SeaweedFileSystem \
|
||||
* --conf spark.hadoop.fs.seaweed.filer.host=localhost \
|
||||
* --conf spark.hadoop.fs.seaweed.filer.port=8888 \
|
||||
* --conf spark.hadoop.fs.seaweed.filer.port.grpc=18888 \
|
||||
* target/seaweedfs-spark-integration-tests-1.0-SNAPSHOT.jar \
|
||||
* seaweedfs://localhost:8888/output
|
||||
*/
|
||||
public class SparkSeaweedFSExample {
|
||||
|
||||
public static void main(String[] args) {
|
||||
if (args.length < 1) {
|
||||
System.err.println("Usage: SparkSeaweedFSExample <output-path>");
|
||||
System.err.println("Example: seaweedfs://localhost:8888/spark-output");
|
||||
System.exit(1);
|
||||
}
|
||||
|
||||
String outputPath = args[0];
|
||||
|
||||
// Create Spark session
|
||||
SparkSession spark = SparkSession.builder()
|
||||
.appName("SeaweedFS Spark Example")
|
||||
.getOrCreate();
|
||||
|
||||
try {
|
||||
System.out.println("=== SeaweedFS Spark Integration Example ===\n");
|
||||
|
||||
// Example 1: Generate data and write to SeaweedFS
|
||||
System.out.println("1. Generating sample data...");
|
||||
Dataset<Row> data = spark.range(0, 1000)
|
||||
.selectExpr(
|
||||
"id",
|
||||
"id * 2 as doubled",
|
||||
"CAST(rand() * 100 AS INT) as random_value"
|
||||
);
|
||||
|
||||
System.out.println(" Generated " + data.count() + " rows");
|
||||
data.show(5);
|
||||
|
||||
// Write as Parquet
|
||||
String parquetPath = outputPath + "/data.parquet";
|
||||
System.out.println("\n2. Writing data to SeaweedFS as Parquet...");
|
||||
System.out.println(" Path: " + parquetPath);
|
||||
|
||||
data.write()
|
||||
.mode(SaveMode.Overwrite)
|
||||
.parquet(parquetPath);
|
||||
|
||||
System.out.println(" ✓ Write completed");
|
||||
|
||||
// Read back and verify
|
||||
System.out.println("\n3. Reading data back from SeaweedFS...");
|
||||
Dataset<Row> readData = spark.read().parquet(parquetPath);
|
||||
System.out.println(" Read " + readData.count() + " rows");
|
||||
|
||||
// Perform aggregation
|
||||
System.out.println("\n4. Performing aggregation...");
|
||||
Dataset<Row> stats = readData.selectExpr(
|
||||
"COUNT(*) as count",
|
||||
"AVG(random_value) as avg_random",
|
||||
"MAX(doubled) as max_doubled"
|
||||
);
|
||||
|
||||
stats.show();
|
||||
|
||||
// Write aggregation results
|
||||
String statsPath = outputPath + "/stats.parquet";
|
||||
System.out.println("5. Writing stats to: " + statsPath);
|
||||
stats.write()
|
||||
.mode(SaveMode.Overwrite)
|
||||
.parquet(statsPath);
|
||||
|
||||
// Create a partitioned dataset
|
||||
System.out.println("\n6. Creating partitioned dataset...");
|
||||
Dataset<Row> partitionedData = data.selectExpr(
|
||||
"*",
|
||||
"CAST(id % 10 AS INT) as partition_key"
|
||||
);
|
||||
|
||||
String partitionedPath = outputPath + "/partitioned.parquet";
|
||||
System.out.println(" Path: " + partitionedPath);
|
||||
|
||||
partitionedData.write()
|
||||
.mode(SaveMode.Overwrite)
|
||||
.partitionBy("partition_key")
|
||||
.parquet(partitionedPath);
|
||||
|
||||
System.out.println(" ✓ Partitioned write completed");
|
||||
|
||||
// Read specific partition
|
||||
System.out.println("\n7. Reading specific partition (partition_key=0)...");
|
||||
Dataset<Row> partition0 = spark.read()
|
||||
.parquet(partitionedPath)
|
||||
.filter("partition_key = 0");
|
||||
|
||||
System.out.println(" Partition 0 contains " + partition0.count() + " rows");
|
||||
partition0.show(5);
|
||||
|
||||
// SQL example
|
||||
System.out.println("\n8. Using Spark SQL...");
|
||||
readData.createOrReplaceTempView("seaweedfs_data");
|
||||
|
||||
Dataset<Row> sqlResult = spark.sql(
|
||||
"SELECT " +
|
||||
" CAST(id / 100 AS INT) as bucket, " +
|
||||
" COUNT(*) as count, " +
|
||||
" AVG(random_value) as avg_random " +
|
||||
"FROM seaweedfs_data " +
|
||||
"GROUP BY CAST(id / 100 AS INT) " +
|
||||
"ORDER BY bucket"
|
||||
);
|
||||
|
||||
System.out.println(" Bucketed statistics:");
|
||||
sqlResult.show();
|
||||
|
||||
System.out.println("\n=== Example completed successfully! ===");
|
||||
System.out.println("Output location: " + outputPath);
|
||||
|
||||
} catch (Exception e) {
|
||||
System.err.println("Error: " + e.getMessage());
|
||||
e.printStackTrace();
|
||||
System.exit(1);
|
||||
} finally {
|
||||
spark.stop();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
package seaweed.spark;
|
||||
|
||||
import org.apache.spark.sql.Dataset;
|
||||
import org.apache.spark.sql.Row;
|
||||
import org.apache.spark.sql.SaveMode;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* Integration tests for Spark read/write operations with SeaweedFS.
|
||||
*/
|
||||
public class SparkReadWriteTest extends SparkTestBase {
|
||||
|
||||
@Test
|
||||
public void testWriteAndReadParquet() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create test data
|
||||
List<Person> people = Arrays.asList(
|
||||
new Person("Alice", 30),
|
||||
new Person("Bob", 25),
|
||||
new Person("Charlie", 35)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(people, Person.class);
|
||||
|
||||
// Write to SeaweedFS
|
||||
String outputPath = getTestPath("people.parquet");
|
||||
df.write().mode(SaveMode.Overwrite).parquet(outputPath);
|
||||
|
||||
// Read back from SeaweedFS
|
||||
Dataset<Row> readDf = spark.read().parquet(outputPath);
|
||||
|
||||
// Verify
|
||||
assertEquals(3, readDf.count());
|
||||
assertEquals(2, readDf.columns().length);
|
||||
|
||||
List<Row> results = readDf.collectAsList();
|
||||
assertTrue(results.stream().anyMatch(r -> "Alice".equals(r.getAs("name")) && (Integer)r.getAs("age") == 30));
|
||||
assertTrue(results.stream().anyMatch(r -> "Bob".equals(r.getAs("name")) && (Integer)r.getAs("age") == 25));
|
||||
assertTrue(results.stream().anyMatch(r -> "Charlie".equals(r.getAs("name")) && (Integer)r.getAs("age") == 35));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWriteAndReadCSV() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create test data
|
||||
List<Person> people = Arrays.asList(
|
||||
new Person("Alice", 30),
|
||||
new Person("Bob", 25)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(people, Person.class);
|
||||
|
||||
// Write to SeaweedFS as CSV
|
||||
String outputPath = getTestPath("people.csv");
|
||||
df.write().mode(SaveMode.Overwrite).option("header", "true").csv(outputPath);
|
||||
|
||||
// Read back from SeaweedFS
|
||||
Dataset<Row> readDf = spark.read().option("header", "true").option("inferSchema", "true").csv(outputPath);
|
||||
|
||||
// Verify
|
||||
assertEquals(2, readDf.count());
|
||||
assertEquals(2, readDf.columns().length);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWriteAndReadJSON() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create test data
|
||||
List<Person> people = Arrays.asList(
|
||||
new Person("Alice", 30),
|
||||
new Person("Bob", 25),
|
||||
new Person("Charlie", 35)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(people, Person.class);
|
||||
|
||||
// Write to SeaweedFS as JSON
|
||||
String outputPath = getTestPath("people.json");
|
||||
df.write().mode(SaveMode.Overwrite).json(outputPath);
|
||||
|
||||
// Read back from SeaweedFS
|
||||
Dataset<Row> readDf = spark.read().json(outputPath);
|
||||
|
||||
// Verify
|
||||
assertEquals(3, readDf.count());
|
||||
assertEquals(2, readDf.columns().length);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWritePartitionedData() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create test data with multiple years
|
||||
List<PersonWithYear> people = Arrays.asList(
|
||||
new PersonWithYear("Alice", 30, 2020),
|
||||
new PersonWithYear("Bob", 25, 2021),
|
||||
new PersonWithYear("Charlie", 35, 2020),
|
||||
new PersonWithYear("David", 28, 2021)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(people, PersonWithYear.class);
|
||||
|
||||
// Write partitioned data to SeaweedFS
|
||||
String outputPath = getTestPath("people_partitioned");
|
||||
df.write().mode(SaveMode.Overwrite).partitionBy("year").parquet(outputPath);
|
||||
|
||||
// Read back from SeaweedFS
|
||||
Dataset<Row> readDf = spark.read().parquet(outputPath);
|
||||
|
||||
// Verify
|
||||
assertEquals(4, readDf.count());
|
||||
|
||||
// Verify partition filtering works
|
||||
Dataset<Row> filtered2020 = readDf.filter("year = 2020");
|
||||
assertEquals(2, filtered2020.count());
|
||||
|
||||
Dataset<Row> filtered2021 = readDf.filter("year = 2021");
|
||||
assertEquals(2, filtered2021.count());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAppendMode() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
String outputPath = getTestPath("people_append.parquet");
|
||||
|
||||
// Write first batch
|
||||
List<Person> batch1 = Arrays.asList(
|
||||
new Person("Alice", 30),
|
||||
new Person("Bob", 25)
|
||||
);
|
||||
Dataset<Row> df1 = spark.createDataFrame(batch1, Person.class);
|
||||
df1.write().mode(SaveMode.Overwrite).parquet(outputPath);
|
||||
|
||||
// Append second batch
|
||||
List<Person> batch2 = Arrays.asList(
|
||||
new Person("Charlie", 35),
|
||||
new Person("David", 28)
|
||||
);
|
||||
Dataset<Row> df2 = spark.createDataFrame(batch2, Person.class);
|
||||
df2.write().mode(SaveMode.Append).parquet(outputPath);
|
||||
|
||||
// Read back and verify
|
||||
Dataset<Row> readDf = spark.read().parquet(outputPath);
|
||||
assertEquals(4, readDf.count());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLargeDataset() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create a larger dataset
|
||||
Dataset<Row> largeDf = spark.range(0, 10000)
|
||||
.selectExpr("id as value", "id * 2 as doubled");
|
||||
|
||||
String outputPath = getTestPath("large_dataset.parquet");
|
||||
largeDf.write().mode(SaveMode.Overwrite).parquet(outputPath);
|
||||
|
||||
// Read back and verify
|
||||
Dataset<Row> readDf = spark.read().parquet(outputPath);
|
||||
assertEquals(10000, readDf.count());
|
||||
|
||||
// Verify some data
|
||||
Row firstRow = readDf.first();
|
||||
assertEquals(0L, firstRow.getLong(0));
|
||||
assertEquals(0L, firstRow.getLong(1));
|
||||
}
|
||||
|
||||
// Test data classes
|
||||
public static class Person implements java.io.Serializable {
|
||||
private String name;
|
||||
private int age;
|
||||
|
||||
public Person() {}
|
||||
|
||||
public Person(String name, int age) {
|
||||
this.name = name;
|
||||
this.age = age;
|
||||
}
|
||||
|
||||
public String getName() { return name; }
|
||||
public void setName(String name) { this.name = name; }
|
||||
public int getAge() { return age; }
|
||||
public void setAge(int age) { this.age = age; }
|
||||
}
|
||||
|
||||
public static class PersonWithYear implements java.io.Serializable {
|
||||
private String name;
|
||||
private int age;
|
||||
private int year;
|
||||
|
||||
public PersonWithYear() {}
|
||||
|
||||
public PersonWithYear(String name, int age, int year) {
|
||||
this.name = name;
|
||||
this.age = age;
|
||||
this.year = year;
|
||||
}
|
||||
|
||||
public String getName() { return name; }
|
||||
public void setName(String name) { this.name = name; }
|
||||
public int getAge() { return age; }
|
||||
public void setAge(int age) { this.age = age; }
|
||||
public int getYear() { return year; }
|
||||
public void setYear(int year) { this.year = year; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
package seaweed.spark;
|
||||
|
||||
import org.apache.spark.sql.Dataset;
|
||||
import org.apache.spark.sql.Row;
|
||||
import org.apache.spark.sql.SaveMode;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* Integration tests for Spark SQL operations with SeaweedFS.
|
||||
*/
|
||||
public class SparkSQLTest extends SparkTestBase {
|
||||
|
||||
@Test
|
||||
public void testCreateTableAndQuery() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create test data
|
||||
List<Employee> employees = Arrays.asList(
|
||||
new Employee(1, "Alice", "Engineering", 100000),
|
||||
new Employee(2, "Bob", "Sales", 80000),
|
||||
new Employee(3, "Charlie", "Engineering", 120000),
|
||||
new Employee(4, "David", "Sales", 75000)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(employees, Employee.class);
|
||||
|
||||
// Write to SeaweedFS
|
||||
String tablePath = getTestPath("employees");
|
||||
df.write().mode(SaveMode.Overwrite).parquet(tablePath);
|
||||
|
||||
// Create temporary view
|
||||
Dataset<Row> employeesDf = spark.read().parquet(tablePath);
|
||||
employeesDf.createOrReplaceTempView("employees");
|
||||
|
||||
// Run SQL queries
|
||||
Dataset<Row> engineeringEmployees = spark.sql(
|
||||
"SELECT name, salary FROM employees WHERE department = 'Engineering'"
|
||||
);
|
||||
|
||||
assertEquals(2, engineeringEmployees.count());
|
||||
|
||||
Dataset<Row> highPaidEmployees = spark.sql(
|
||||
"SELECT name, salary FROM employees WHERE salary > 90000"
|
||||
);
|
||||
|
||||
assertEquals(2, highPaidEmployees.count());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAggregationQueries() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create sales data
|
||||
List<Sale> sales = Arrays.asList(
|
||||
new Sale("2024-01", "Product A", 100),
|
||||
new Sale("2024-01", "Product B", 150),
|
||||
new Sale("2024-02", "Product A", 120),
|
||||
new Sale("2024-02", "Product B", 180),
|
||||
new Sale("2024-03", "Product A", 110)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(sales, Sale.class);
|
||||
|
||||
// Write to SeaweedFS
|
||||
String tablePath = getTestPath("sales");
|
||||
df.write().mode(SaveMode.Overwrite).parquet(tablePath);
|
||||
|
||||
// Create temporary view
|
||||
Dataset<Row> salesDf = spark.read().parquet(tablePath);
|
||||
salesDf.createOrReplaceTempView("sales");
|
||||
|
||||
// Aggregate query
|
||||
Dataset<Row> monthlySales = spark.sql(
|
||||
"SELECT month, SUM(amount) as total FROM sales GROUP BY month ORDER BY month"
|
||||
);
|
||||
|
||||
List<Row> results = monthlySales.collectAsList();
|
||||
assertEquals(3, results.size());
|
||||
assertEquals("2024-01", results.get(0).getString(0));
|
||||
assertEquals(250, results.get(0).getLong(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testJoinOperations() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create employee data
|
||||
List<Employee> employees = Arrays.asList(
|
||||
new Employee(1, "Alice", "Engineering", 100000),
|
||||
new Employee(2, "Bob", "Sales", 80000)
|
||||
);
|
||||
|
||||
// Create department data
|
||||
List<Department> departments = Arrays.asList(
|
||||
new Department("Engineering", "Building Products"),
|
||||
new Department("Sales", "Selling Products")
|
||||
);
|
||||
|
||||
Dataset<Row> empDf = spark.createDataFrame(employees, Employee.class);
|
||||
Dataset<Row> deptDf = spark.createDataFrame(departments, Department.class);
|
||||
|
||||
// Write to SeaweedFS
|
||||
String empPath = getTestPath("employees_join");
|
||||
String deptPath = getTestPath("departments_join");
|
||||
|
||||
empDf.write().mode(SaveMode.Overwrite).parquet(empPath);
|
||||
deptDf.write().mode(SaveMode.Overwrite).parquet(deptPath);
|
||||
|
||||
// Read back and create views
|
||||
spark.read().parquet(empPath).createOrReplaceTempView("emp");
|
||||
spark.read().parquet(deptPath).createOrReplaceTempView("dept");
|
||||
|
||||
// Join query
|
||||
Dataset<Row> joined = spark.sql(
|
||||
"SELECT e.name, e.salary, d.description " +
|
||||
"FROM emp e JOIN dept d ON e.department = d.name"
|
||||
);
|
||||
|
||||
assertEquals(2, joined.count());
|
||||
|
||||
List<Row> results = joined.collectAsList();
|
||||
assertTrue(results.stream().anyMatch(r ->
|
||||
"Alice".equals(r.getString(0)) && "Building Products".equals(r.getString(2))
|
||||
));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWindowFunctions() {
|
||||
skipIfTestsDisabled();
|
||||
|
||||
// Create employee data with salaries
|
||||
List<Employee> employees = Arrays.asList(
|
||||
new Employee(1, "Alice", "Engineering", 100000),
|
||||
new Employee(2, "Bob", "Engineering", 120000),
|
||||
new Employee(3, "Charlie", "Sales", 80000),
|
||||
new Employee(4, "David", "Sales", 90000)
|
||||
);
|
||||
|
||||
Dataset<Row> df = spark.createDataFrame(employees, Employee.class);
|
||||
|
||||
String tablePath = getTestPath("employees_window");
|
||||
df.write().mode(SaveMode.Overwrite).parquet(tablePath);
|
||||
|
||||
Dataset<Row> employeesDf = spark.read().parquet(tablePath);
|
||||
employeesDf.createOrReplaceTempView("employees_ranked");
|
||||
|
||||
// Window function query - rank employees by salary within department
|
||||
Dataset<Row> ranked = spark.sql(
|
||||
"SELECT name, department, salary, " +
|
||||
"RANK() OVER (PARTITION BY department ORDER BY salary DESC) as rank " +
|
||||
"FROM employees_ranked"
|
||||
);
|
||||
|
||||
assertEquals(4, ranked.count());
|
||||
|
||||
// Verify Bob has rank 1 in Engineering (highest salary)
|
||||
List<Row> results = ranked.collectAsList();
|
||||
Row bobRow = results.stream()
|
||||
.filter(r -> "Bob".equals(r.getString(0)))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
|
||||
assertNotNull(bobRow);
|
||||
assertEquals(1, bobRow.getInt(3));
|
||||
}
|
||||
|
||||
// Test data classes
|
||||
public static class Employee implements java.io.Serializable {
|
||||
private int id;
|
||||
private String name;
|
||||
private String department;
|
||||
private int salary;
|
||||
|
||||
public Employee() {}
|
||||
|
||||
public Employee(int id, String name, String department, int salary) {
|
||||
this.id = id;
|
||||
this.name = name;
|
||||
this.department = department;
|
||||
this.salary = salary;
|
||||
}
|
||||
|
||||
public int getId() { return id; }
|
||||
public void setId(int id) { this.id = id; }
|
||||
public String getName() { return name; }
|
||||
public void setName(String name) { this.name = name; }
|
||||
public String getDepartment() { return department; }
|
||||
public void setDepartment(String department) { this.department = department; }
|
||||
public int getSalary() { return salary; }
|
||||
public void setSalary(int salary) { this.salary = salary; }
|
||||
}
|
||||
|
||||
public static class Sale implements java.io.Serializable {
|
||||
private String month;
|
||||
private String product;
|
||||
private int amount;
|
||||
|
||||
public Sale() {}
|
||||
|
||||
public Sale(String month, String product, int amount) {
|
||||
this.month = month;
|
||||
this.product = product;
|
||||
this.amount = amount;
|
||||
}
|
||||
|
||||
public String getMonth() { return month; }
|
||||
public void setMonth(String month) { this.month = month; }
|
||||
public String getProduct() { return product; }
|
||||
public void setProduct(String product) { this.product = product; }
|
||||
public int getAmount() { return amount; }
|
||||
public void setAmount(int amount) { this.amount = amount; }
|
||||
}
|
||||
|
||||
public static class Department implements java.io.Serializable {
|
||||
private String name;
|
||||
private String description;
|
||||
|
||||
public Department() {}
|
||||
|
||||
public Department(String name, String description) {
|
||||
this.name = name;
|
||||
this.description = description;
|
||||
}
|
||||
|
||||
public String getName() { return name; }
|
||||
public void setName(String name) { this.name = name; }
|
||||
public String getDescription() { return description; }
|
||||
public void setDescription(String description) { this.description = description; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
package seaweed.spark;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.spark.SparkConf;
|
||||
import org.apache.spark.sql.SparkSession;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* Base class for Spark integration tests with SeaweedFS.
|
||||
*
|
||||
* These tests require a running SeaweedFS cluster.
|
||||
* Set environment variable SEAWEEDFS_TEST_ENABLED=true to enable these tests.
|
||||
*/
|
||||
public abstract class SparkTestBase {
|
||||
|
||||
protected SparkSession spark;
|
||||
protected static final String TEST_ROOT = "/test-spark";
|
||||
protected static final boolean TESTS_ENABLED =
|
||||
"true".equalsIgnoreCase(System.getenv("SEAWEEDFS_TEST_ENABLED"));
|
||||
|
||||
// SeaweedFS connection settings
|
||||
protected static final String SEAWEEDFS_HOST =
|
||||
System.getenv().getOrDefault("SEAWEEDFS_FILER_HOST", "localhost");
|
||||
protected static final String SEAWEEDFS_PORT =
|
||||
System.getenv().getOrDefault("SEAWEEDFS_FILER_PORT", "8888");
|
||||
protected static final String SEAWEEDFS_GRPC_PORT =
|
||||
System.getenv().getOrDefault("SEAWEEDFS_FILER_GRPC_PORT", "18888");
|
||||
|
||||
@Before
|
||||
public void setUpSpark() throws IOException {
|
||||
if (!TESTS_ENABLED) {
|
||||
return;
|
||||
}
|
||||
|
||||
SparkConf sparkConf = new SparkConf()
|
||||
.setAppName("SeaweedFS Integration Test")
|
||||
.setMaster("local[1]") // Single thread to avoid concurrent gRPC issues
|
||||
.set("spark.driver.host", "localhost")
|
||||
.set("spark.sql.warehouse.dir", getSeaweedFSPath("/spark-warehouse"))
|
||||
// SeaweedFS configuration
|
||||
.set("spark.hadoop.fs.defaultFS", String.format("seaweedfs://%s:%s", SEAWEEDFS_HOST, SEAWEEDFS_PORT))
|
||||
.set("spark.hadoop.fs.seaweedfs.impl", "seaweed.hdfs.SeaweedFileSystem")
|
||||
.set("spark.hadoop.fs.seaweed.impl", "seaweed.hdfs.SeaweedFileSystem")
|
||||
.set("spark.hadoop.fs.seaweed.filer.host", SEAWEEDFS_HOST)
|
||||
.set("spark.hadoop.fs.seaweed.filer.port", SEAWEEDFS_PORT)
|
||||
.set("spark.hadoop.fs.seaweed.filer.port.grpc", SEAWEEDFS_GRPC_PORT)
|
||||
.set("spark.hadoop.fs.AbstractFileSystem.seaweedfs.impl", "seaweed.hdfs.SeaweedAbstractFileSystem")
|
||||
// Set replication to empty string to use filer default
|
||||
.set("spark.hadoop.fs.seaweed.replication", "")
|
||||
// Smaller buffer to reduce load
|
||||
.set("spark.hadoop.fs.seaweed.buffer.size", "1048576") // 1MB
|
||||
// Reduce parallelism
|
||||
.set("spark.default.parallelism", "1")
|
||||
.set("spark.sql.shuffle.partitions", "1")
|
||||
// Simpler output committer
|
||||
.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")
|
||||
.set("spark.sql.sources.commitProtocolClass", "org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol")
|
||||
// Disable speculative execution to reduce load
|
||||
.set("spark.speculation", "false");
|
||||
|
||||
spark = SparkSession.builder()
|
||||
.config(sparkConf)
|
||||
.getOrCreate();
|
||||
|
||||
// Clean up test directory
|
||||
cleanupTestDirectory();
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDownSpark() {
|
||||
if (!TESTS_ENABLED || spark == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// Try to cleanup but don't fail if it doesn't work
|
||||
cleanupTestDirectory();
|
||||
} catch (Exception e) {
|
||||
System.err.println("Cleanup failed: " + e.getMessage());
|
||||
} finally {
|
||||
try {
|
||||
spark.stop();
|
||||
} catch (Exception e) {
|
||||
System.err.println("Spark stop failed: " + e.getMessage());
|
||||
}
|
||||
spark = null;
|
||||
}
|
||||
}
|
||||
|
||||
protected String getSeaweedFSPath(String path) {
|
||||
return String.format("seaweedfs://%s:%s%s", SEAWEEDFS_HOST, SEAWEEDFS_PORT, path);
|
||||
}
|
||||
|
||||
protected String getTestPath(String subPath) {
|
||||
return getSeaweedFSPath(TEST_ROOT + "/" + subPath);
|
||||
}
|
||||
|
||||
private void cleanupTestDirectory() {
|
||||
if (spark != null) {
|
||||
try {
|
||||
Configuration conf = spark.sparkContext().hadoopConfiguration();
|
||||
org.apache.hadoop.fs.FileSystem fs = org.apache.hadoop.fs.FileSystem.get(
|
||||
java.net.URI.create(getSeaweedFSPath("/")), conf);
|
||||
org.apache.hadoop.fs.Path testPath = new org.apache.hadoop.fs.Path(TEST_ROOT);
|
||||
if (fs.exists(testPath)) {
|
||||
fs.delete(testPath, true);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// Suppress cleanup errors - they shouldn't fail tests
|
||||
// Common in distributed systems with eventual consistency
|
||||
System.err.println("Warning: cleanup failed (non-critical): " + e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected void skipIfTestsDisabled() {
|
||||
if (!TESTS_ENABLED) {
|
||||
System.out.println("Skipping test - SEAWEEDFS_TEST_ENABLED not set");
|
||||
org.junit.Assume.assumeTrue("SEAWEEDFS_TEST_ENABLED not set", false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
# Set root logger level
|
||||
log4j.rootLogger=WARN, console
|
||||
|
||||
# Console appender
|
||||
log4j.appender.console=org.apache.log4j.ConsoleAppender
|
||||
log4j.appender.console.target=System.err
|
||||
log4j.appender.console.layout=org.apache.log4j.PatternLayout
|
||||
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
|
||||
|
||||
# Set log levels for specific packages
|
||||
log4j.logger.org.apache.spark=WARN
|
||||
log4j.logger.org.apache.hadoop=WARN
|
||||
log4j.logger.seaweed=INFO
|
||||
|
||||
# Suppress unnecessary warnings
|
||||
log4j.logger.org.apache.spark.util.Utils=ERROR
|
||||
log4j.logger.org.apache.hadoop.util.NativeCodeLoader=ERROR
|
||||
|
||||
Reference in New Issue
Block a user