# Spark Java Demo 示例

以下是使用 Java 编写的 Apache Spark 示例，展示如何进行基本的数据处理操作。

## 1. 基础 WordCount 示例 (Java)

```java
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

public class JavaWordCount {
    public static void main(String[] args) {
        // 创建 SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("Java WordCount")
                .master("local[*]")
                .getOrCreate();

        // 获取 JavaSparkContext
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        // 示例数据
        List<String> data = Arrays.asList(
                "Hello Spark",
                "Hello World",
                "Spark is awesome"
        );

        // 创建 RDD
        JavaRDD<String> rdd = jsc.parallelize(data);

        // 执行 WordCount
        JavaPairRDD<String, Integer> wordCounts = rdd
                .flatMap(line -> Arrays.asList(line.split(" ")).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1))
                .reduceByKey(Integer::sum);

        // 收集并打印结果
        for (Tuple2<String, Integer> tuple : wordCounts.collect()) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }

        // 停止 SparkSession
        spark.stop();
    }
}
```

## 2. 使用 DataFrame API 的示例 (Java)

```java
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.Arrays;
import java.util.List;

public class JavaDataFrameDemo {
    public static void main(String[] args) {
        // 创建 SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("Java DataFrame Demo")
                .master("local[*]")
                .getOrCreate();

        // 示例数据
        List<Row> data = Arrays.asList(
                RowFactory.create("Alice", 25),
                RowFactory.create("Bob", 30),
                RowFactory.create("Charlie", 35)
        );

        // 定义 schema
        StructType schema = new StructType(new StructField[]{
                new StructField("Name", DataTypes.StringType, false, null),
                new StructField("Age", DataTypes.IntegerType, false, null)
        });

        // 创建 DataFrame
        Dataset<Row> df = spark.createDataFrame(data, schema);

        // 显示 DataFrame
        df.show();

        // 过滤年龄大于30的人
        Dataset<Row> filteredDF = df.filter("Age > 30");
        filteredDF.show();

        // 按年龄分组求和
        Dataset<Row> groupedDF = df.groupBy().sum("Age");
        groupedDF.show();

        // 停止 SparkSession
        spark.stop();
    }
}
```

## 3. 从文件读取数据的示例 (Java)

```java
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class JavaFileReadingDemo {
    public static void main(String[] args) {
        // 创建 SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("Java File Reading Demo")
                .master("local[*]")
                .getOrCreate();

        // 从 CSV 文件读取数据 (假设文件在本地或HDFS上)
        // Dataset<Row> df = spark.read()
        //         .option("header", "true")
        //         .option("inferSchema", "true")
        //         .csv("path/to/file.csv");

        // 使用示例数据创建临时 DataFrame
        Dataset<Row> df = spark.createDataset(
                Arrays.asList(
                        RowFactory.create("Alice", 25),
                        RowFactory.create("Bob", 30),
                        RowFactory.create("Charlie", 35)
                ),
                RowEncoder.apply(new StructType(new StructField[]{
                        new StructField("Name", DataTypes.StringType, false, null),
                        new StructField("Age", DataTypes.IntegerType, false, null)
                }))
        );

        // 显示 DataFrame 内容
        df.show();

        // 将 DataFrame 写入 CSV 文件
        // df.write()
        //    .option("header", "true")
        //    .csv("output/path");

        // 停止 SparkSession
        spark.stop();
    }
}
```

## 运行 Spark Java Demo 的步骤

1. **环境准备**：
   - 安装 Java JDK 8+
   - 下载并安装 Spark (https://spark.apache.org/downloads.html)
   - 设置 `SPARK_HOME` 环境变量

2. **项目配置**：
   - 使用 Maven 或 Gradle 管理依赖
   - Maven 依赖示例：
     ```xml
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-core_2.12</artifactId>
         <version>3.5.0</version>
     </dependency>
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-sql_2.12</artifactId>
         <version>3.5.0</version>
     </dependency>
     ```

3. **编译和运行**：
   - 编译 Java 文件：`javac -cp "spark_jars/*" JavaWordCount.java`
   - 运行程序：`spark-submit --class JavaWordCount target/your-app.jar`

4. **打包运行**：
   - 使用 Maven 创建 uber jar：`mvn package`
   - 提交运行：`spark-submit --class JavaWordCount target/your-app-with-dependencies.jar`

## 常见问题解决 (Java 版本)

1. **类找不到错误**：
   - 确保所有 Spark 依赖项在 classpath 中
   - 使用 `--jars` 参数指定额外依赖

2. **序列化错误**：
   - 确保使用的函数是可序列化的
   - 对于 lambda 表达式，使用静态方法或实例方法替代

3. **类型转换问题**：
   - Java API 比 Scala/Python 更严格，注意类型转换
   - 使用 `RowFactory` 和 `StructType` 明确定义 schema

需要更具体的 Java Spark 示例或针对特定场景的演示吗？