深入浅出：Eclipse 中配置 Maven 与 Spark 应用开发全指南

Spark

安装配置

1.在 Eclipse 中配置 Maven

    Eclipse 中默认自带 Maven 插件，但是自带的 Maven 插件不能修改本地仓库，所

以通常我们不使用自带的 Maven ，而是使用自己安装的，在 Eclipse 中配置 Maven 的

步骤如下：

点击 Eclipse 中的 Window → Preferences
点开 Maven 前面的箭头，选择 Installations，点击 Add…

点击 Directory…选择我们安装的 Maven 核心程序的根目录，然后点击 Finish

勾上添加的 Maven 核心程序
选择 Maven 下的 User Settings ，在全局设置哪儿点击 Browse… 选择 Maven

核心程序的配置文件 settings.xml ，本地仓库会自动变为我们在 settings.xml

文件中设置的路径

修改setting内容：

<profile>

      <id>jdk-1.8</id>

        <activation>

                 <activeByDefault>true</activeByDefault>

                <jdk>1.8</jdk>

        </activation>

<properties>

<maven.compiler.source>1.8</maven.compiler.source>

<maven.compiler.target>1.8</maven.compiler.target>

<maven.compiler.compilerversion>1.8</maven.compiler.compilerversion>

</properties>

</profile>

在 Eclipse 中创建 Maven 项目

2.1 创建 Java 工程

点击 File → New → Maven Project ，弹出如下窗口
点击 Next，配置坐标（GAV）及打包方式，然后点击 Finish

group id：组织id

artifact id：项目名字

version：版本

package：java包名

创建成功后，配置 Maven 的核心配置文件 pom.xml 文件

<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>me.spark.app</groupId>

  <artifactId>playersStats</artifactId>

  <version>1.0</version>

  <name>playersStats</name>

  <!-- FIXME change it to the project's website -->

  <url>http://www.example.com</url>

  <properties>

    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

    <maven.compiler.source>1.8</maven.compiler.source>

    <maven.compiler.target>1.8</maven.compiler.target>

  </properties>

  <dependencies>

    <!-- https://mvnrepository.com/artifact/org.apache.maven.plugins/maven-assembly-plugin -->

        <dependency>

      <groupId>org.apache.maven.plugins</groupId>

      <artifactId>maven-assembly-plugin</artifactId>

      <version>3.3.0</version>

    </dependency>

      <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->

      <dependency>

      <groupId>org.apache.spark</groupId>

      <artifactId>spark-core_2.12</artifactId>

      <version>3.0.0</version>

    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->

        <dependency>

      <groupId>org.apache.spark</groupId>

      <artifactId>spark-sql_2.12</artifactId>

      <version>3.0.0</version>

    </dependency>

  </dependencies>

  <build>

        <plugins>

        <plugin>

            <artifactId>maven-assembly-plugin</artifactId>

            <version>3.3.0</version>

            <configuration>

          <descriptorRefs>

            <descriptorRef>jar-with-dependencies</descriptorRef>

          </descriptorRefs>

        <archive>

          <manifest>

            <mainClass>me.spark.app.playersStats.Main</mainClass>

          </manifest>

        </archive>

        </configuration>

        <executions>

            <execution>

                <id>make-assembly</id> <!-- this is used for inheritance merges -->

                <phase>package</phase> <!-- bind to the packaging phase -->

                <goals>

                <goal>single</goal>

                </goals>

            </execution>

        </executions>

      </plugin>

    </plugins>

  </build>

</project>

模板

1. Spark 应用基本模板

这是 Spark 应用程序的基本结构，适用于所有 Spark 程序的起点。

import org.apache.spark.sql.SparkSession;

public class MySparkApp {

    public static void main(String[] args) {

        // 创建 SparkSession，应用程序入口

        SparkSession spark = SparkSession.builder()

                .appName("MySparkApp")

                .master("local")  // 本地模式

                .getOrCreate();

        // 你的 Spark 代码在这里编写

        // 关闭 SparkSession

        spark.stop();

    }

}

2. DataFrame 操作模板

DataFrame 是 Spark 进行结构化数据处理的核心 API，特别适合处理 CSV、JSON、Parquet 等格式的文件。以下是通过 DataFrame 进行数据处理的模板。

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 MyDataFrameApp {

    public static void main(String[] args) {

        // 创建 SparkSession

        SparkSession spark = SparkSession.builder()

                .appName("DataFrameExample")

                .master("local")

                .getOrCreate();

        // 读取 CSV 文件为 DataFrame

        Dataset<Row> df = spark.read()

                .option("header", "true") // 是否包含头部

                .option("inferSchema", "true") // 自动推断列类型

                .csv("path/to/your/csvfile.csv");

        // 打印 DataFrame 的结构

        df.printSchema();

       

        // 显示前20行数据

        df.show();

        // 数据处理：计算某列的平均值

        df.groupBy("column_name")

          .agg(avg("another_column").alias("average_value"))

          .show();

        // 关闭 SparkSession

        spark.stop();

    }

}

3. RDD 操作模板

RDD（Resilient Distributed Dataset）是 Spark 的底层 API，适合处理非结构化数据，特别是在数据量较大时。以下是通过 RDD 进行数据处理的模板。

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.SparkConf;

public class MyRDDApp {

    public static void main(String[] args) {

        // 配置 Spark

        SparkConf conf = new SparkConf().setAppName("RDDExample").setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 读取文本文件为 RDD

        JavaRDD<String> lines = sc.textFile("path/to/your/file.txt");

        // 简单的 WordCount 示例

        JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());

        JavaRDD<Tuple2<String, Integer>> wordCounts = words.mapToPair(word -> new Tuple2<>(word, 1))

                .reduceByKey((a, b) -> a + b);

        // 打印结果

        wordCounts.collect().forEach(System.out::println);

        // 关闭 SparkContext

        sc.close();

    }

}

4. Spark SQL 模板

Spark SQL 允许你使用 SQL 查询来处理结构化数据。以下是 Spark SQL 的使用模板，适合数据分析任务。

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

public class MySQLApp {

    public static void main(String[] args) {

        // 创建 SparkSession

        SparkSession spark = SparkSession.builder()

                .appName("SQLExample")

                .master("local")

                .getOrCreate();

        // 读取 CSV 文件为 DataFrame

        Dataset<Row> df = spark.read()

                .option("header", "true")

                .option("inferSchema", "true")

                .csv("path/to/your/csvfile.csv");

        // 注册临时表

        df.createOrReplaceTempView("my_table");

        // 使用 SQL 查询数据

        Dataset<Row> result = spark.sql("SELECT column_name, COUNT(*) FROM my_table GROUP BY column_name");

        // 显示查询结果

        result.show();

        // 关闭 SparkSession

        spark.stop();

    }

}

5. 数据读取与写入模板

Spark 支持多种数据源的读取和写入，如 CSV、JSON、Parquet、JDBC 等。以下是常见的读取和写入数据的操作模板。

读取 CSV 数据

Dataset<Row> df = spark.read()

        .option("header", "true")

        .option("inferSchema", "true")

        .csv("path/to/csvfile.csv");

读取 JSON 数据

Dataset<Row> df = spark.read()

        .json("path/to/jsonfile.json");

读取 Parquet 数据

Dataset<Row> df = spark.read()

        .parquet("path/to/parquetfile.parquet");

写入数据到 CSV 文件

df.write()

  .option("header", "true")

  .csv("path/to/output_csv/");

写入数据到 Parquet 文件

df.write()

  .parquet("path/to/output_parquet/");

6. Spark Streaming 模板

Spark Streaming 处理实时流数据。以下是通过 Spark Streaming 进行数据处理的模板。

import org.apache.spark.SparkConf;

import org.apache.spark.streaming.Durations;

import org.apache.spark.streaming.api.java.JavaStreamingContext;

public class MyStreamingApp {

    public static void main(String[] args) throws InterruptedException {

        // 配置 Spark Streaming

        SparkConf conf = new SparkConf().setAppName("StreamingExample").setMaster("local[2]");

        JavaStreamingContext streamingContext = new JavaStreamingContext(conf, Durations.seconds(1));

        // 监听 socket 数据源

        JavaReceiverInputDStream<String> lines = streamingContext.socketTextStream("localhost", 9999);

        // 处理数据：简单的 WordCount

        JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());

        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(word -> new Tuple2<>(word, 1))

                .reduceByKey((a, b) -> a + b);

        // 输出结果

        wordCounts.print();

        // 启动流处理

        streamingContext.start();

        streamingContext.awaitTermination();

    }

}

标签：大数据 spark maven

本文转载自: https://blog.csdn.net/m0_74070923/article/details/142439365
版权归原作者 小伍_Five 所有，如有侵权，请联系我们删除。