mkdir spark-quickstart
cd spark-quickstart

plugins {
    id 'java'
    id 'application'
}

group = 'com.example'
version = '1.0.0'

java {
    sourceCompatibility = '17'
}

repositories {
    mavenCentral()
}

dependencies {
    // Spark Core
    implementation 'org.apache.spark:spark-core_2.13:3.5.1'
    // Spark SQL (DataFrame, Dataset 사용을 위해)
    implementation 'org.apache.spark:spark-sql_2.13:3.5.1'

    // 로깅
    implementation 'org.slf4j:slf4j-simple:2.0.9'
}

application {
    mainClass = 'com.example.SparkQuickStart'
}

// Spark JAR 충돌 방지
configurations.all {
    exclude group: 'org.slf4j', module: 'slf4j-log4j12'
}

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>spark-quickstart</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <spark.version>3.5.1</spark.version>
        <scala.version>2.13</scala.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.9</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <version>3.1.0</version>
                <configuration>
                    <mainClass>com.example.SparkQuickStart</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

mvn compile exec:java

id,name,department,salary
1,김철수,Engineering,5000
2,이영희,Marketing,4500
3,박민수,Engineering,5500
4,정수진,Sales,4000
5,최동욱,Engineering,6000
6,한미영,Marketing,4800
7,강준혁,Sales,4200
8,윤서연,Engineering,5200

package com.example;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

public class SparkQuickStart {

    public static void main(String[] args) {
        // 1. SparkSession 생성 - Spark의 진입점
        SparkSession spark = SparkSession.builder()
                .appName("Quick Start")
                .master("local[*]")  // 로컬 모드, 모든 코어 사용
                .getOrCreate();

        // 로그 레벨 조정 (너무 많은 로그 방지)
        spark.sparkContext().setLogLevel("WARN");

        System.out.println("=== Spark Quick Start ===\n");

        // 2. CSV 파일 읽기
        Dataset<Row> employees = spark.read()
                .option("header", "true")       // 첫 줄을 헤더로
                .option("inferSchema", "true")  // 타입 자동 추론
                .csv("src/main/resources/employees.csv");

        // 3. 데이터 확인
        System.out.println("=== 전체 직원 데이터 ===");
        employees.show();

        // 4. 스키마 확인
        System.out.println("=== 스키마 ===");
        employees.printSchema();

        // 5. 필터링 - 연봉 5000 이상
        System.out.println("=== 연봉 5000 이상 직원 ===");
        employees.filter(col("salary").geq(5000)).show();

        // 6. 집계 - 부서별 평균 연봉
        System.out.println("=== 부서별 평균 연봉 ===");
        employees.groupBy("department")
                .agg(
                    avg("salary").alias("avg_salary"),
                    count("*").alias("employee_count")
                )
                .orderBy(desc("avg_salary"))
                .show();

        // 7. SQL 사용 - 동일한 작업을 SQL로
        employees.createOrReplaceTempView("employees");

        System.out.println("=== SQL로 조회: Engineering 부서 ===");
        spark.sql("""
            SELECT name, salary
            FROM employees
            WHERE department = 'Engineering'
            ORDER BY salary DESC
            """).show();

        // 8. SparkSession 종료
        spark.stop();

        System.out.println("=== 완료 ===");
    }
}

./gradlew run

gradlew.bat run

package com.example;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.AnalysisException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.spark.sql.functions.*;

public class SparkQuickStartProduction {
    private static final Logger logger = LoggerFactory.getLogger(SparkQuickStartProduction.class);

    public static void main(String[] args) {
        SparkSession spark = null;
        int exitCode = 0;

        try {
            // SparkSession 생성
            spark = SparkSession.builder()
                    .appName("Quick Start Production")
                    .master("local[*]")
                    .config("spark.sql.session.timeZone", "Asia/Seoul")
                    .getOrCreate();

            spark.sparkContext().setLogLevel("WARN");
            logger.info("SparkSession 생성 완료");

            // 데이터 읽기 (에러 모드 설정)
            Dataset<Row> employees = spark.read()
                    .option("header", "true")
                    .option("inferSchema", "true")
                    .option("mode", "FAILFAST")  // 잘못된 데이터 시 즉시 실패
                    .csv("src/main/resources/employees.csv");

            // 스키마 검증
            validateSchema(employees);

            // 데이터 처리
            Dataset<Row> result = employees
                    .filter(col("salary").isNotNull())
                    .groupBy("department")
                    .agg(
                        avg("salary").alias("avg_salary"),
                        count("*").alias("count")
                    );

            // 결과 출력
            result.show();
            logger.info("처리 완료: {} 개 부서", result.count());

        } catch (AnalysisException e) {
            logger.error("데이터 분석 오류: {}", e.getMessage());
            exitCode = 1;
        } catch (Exception e) {
            logger.error("Spark 작업 실패", e);
            exitCode = 1;
        } finally {
            // 리소스 정리 (항상 실행)
            if (spark != null) {
                spark.stop();
                logger.info("SparkSession 종료");
            }
        }

        System.exit(exitCode);
    }

    private static void validateSchema(Dataset<Row> df) {
        String[] requiredColumns = {"id", "name", "department", "salary"};
        for (String col : requiredColumns) {
            if (!java.util.Arrays.asList(df.columns()).contains(col)) {
                throw new IllegalArgumentException("필수 컬럼 누락: " + col);
            }
        }
    }
}

SparkSession spark = SparkSession.builder()
        .appName("Quick Start")
        .master("local[*]")
        .getOrCreate();

Dataset<Row> employees = spark.read()
        .option("header", "true")
        .option("inferSchema", "true")
        .csv("src/main/resources/employees.csv");

employees.filter(col("salary").geq(5000))

employees.createOrReplaceTempView("employees");
spark.sql("SELECT * FROM employees WHERE ...");

// Java Stream (단일 JVM)
List<Employee> highEarners = employees.stream()
        .filter(e -> e.getSalary() >= 5000)
        .collect(Collectors.toList());

// Spark DataFrame (분산 처리)
Dataset<Row> highEarners = employees
        .filter(col("salary").geq(5000));

spark.sparkContext().setLogLevel("WARN");  // 또는 "ERROR"

./gradlew run -Dorg.gradle.jvmargs="-Xmx2g"

application {
    applicationDefaultJvmArgs = ['-Xmx2g']
}