Spark Java POC：CSV数据框双字段逻辑处理技术问询

Hey there! Since you're new to Apache Spark and already have your CSV loaded into a DataFrame, let's break down how to implement data processing logic based on two fields. I'll walk you through common real-world scenarios with Java code examples that align with your goal of replacing your existing batch system.

Adjust the column names below to match your actual dataset—let's use relatable examples like department/performance or age/salary to demonstrate:

1. Filter Rows Based on Dual Conditions #

If you need to keep only rows that meet two criteria (e.g., senior engineering staff with top performance), use either the DataFrame API or Spark SQL:

// Using DataFrame API syntax
Dataset<Row> filteredDf = df.filter(col("department").equalTo("Engineering")
    .and(col("years_of_service").gt(5)));

// Or use SQL if you're more comfortable with it
df.createOrReplaceTempView("employee_data");
Dataset<Row> filteredDfSql = sparkSession.sql(
    "SELECT * FROM employee_data WHERE department = 'Engineering' AND years_of_service > 5"
);

2. Calculate a New Field Using Two Existing Fields #

Suppose you want to compute a bonus based on salary and performance rating:

Dataset<Row> withBonusDf = df.withColumn("bonus", 
    when(col("performance").equalTo("excellent"), col("salary").multiply(0.05))
    .when(col("performance").equalTo("good"), col("salary").multiply(0.03))
    .otherwise(col("salary").multiply(0.01))
);

3. Update an Existing Field Based on Two Conditions #

If you need to adjust values (like raising salaries for eligible employees), use withColumn() to overwrite the existing field:

Dataset<Row> updatedSalaryDf = df.withColumn("salary",
    when(col("department").equalTo("Sales").and(col("quarterly_target_met").equalTo(true)), 
         col("salary").multiply(1.10)) // 10% raise for sales staff who hit targets
    .otherwise(col("salary"))
);

4. Aggregate Data by Two Fields #

For summary tasks (like average salary per department and performance tier), use groupBy() with aggregations:

Dataset<Row> aggregatedDf = df.groupBy("department", "performance")
    .agg(avg("salary").alias("average_salary"), count("*").alias("employee_count"));

Your current batch system reads from database tables—here's how to connect Spark directly to your database (works for MySQL, PostgreSQL, SQL Server, etc.) to read source data or write processed results:

Reading from a Database #

Dataset<Row> dbSourceDf = sparkSession.read()
    .format("jdbc")
    .option("url", "jdbc:mysql://your-db-host:3306/your-database-name")
    .option("dbtable", "your-source-table")
    .option("user", "your-db-username")
    .option("password", "your-db-password")
    .load();

Writing Processed Data Back to Database #

After processing, save your results back to the database (choose the SaveMode that fits your needs):

updatedSalaryDf.write()
    .format("jdbc")
    .option("url", "jdbc:mysql://your-db-host:3306/your-database-name")
    .option("dbtable", "your-target-table")
    .option("user", "your-db-username")
    .option("password", "your-db-password")
    .mode(SaveMode.Overwrite) // Use SaveMode.Append if you don't want to replace existing data
    .save();

• Run df.printSchema() right after loading data to confirm column types (critical when using inferSchema, which can guess wrong sometimes).
• Use df.show(20) to spot-check your data before and after processing—this helps catch mistakes early.
• For super complex logic, split it into smaller DataFrame steps and test each one individually instead of trying to do everything at once.

内容的提问来源于stack exchange，提问作者Curious one