Spark的一些高级用法

Java 中实现 Spark 的一些高级用法。

1. 使用 DataFrame 和 Spark SQL

在 Spark 中，使用 DataFrame 来处理结构化数据并执行 SQL 查询是非常常见的。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSQLExample {public static void main(String[] args) {SparkSession spark = SparkSession.builder().appName("Spark SQL Example").master("local[*]").getOrCreate();// 读取 JSON 文件为 DataFrameDataset<Row> df = spark.read().json("hdfs://path/to/json");// 注册 DataFrame 为临时视图df.createOrReplaceTempView("people");// 使用 Spark SQL 查询Dataset<Row> result = spark.sql("SELECT name, age FROM people WHERE age > 25");// 显示结果result.show();// 显示物理和逻辑执行计划result.explain(true);spark.stop();}
}

2. 缓存和持久化 (Caching and Persistence)

在频繁使用某个 DataFrame 时，可以通过缓存提高性能。这里是 Java 代码示例：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkCachingExample {public static void main(String[] args) {SparkSession spark = SparkSession.builder().appName("Caching Example").master("local[*]").getOrCreate();// 读取 Parquet 文件为 DataFrameDataset<Row> df = spark.read().parquet("hdfs://path/to/parquet");// 缓存 DataFramedf.cache();// 对缓存数据进行操作long count = df.filter(df.col("age").gt(30)).count();System.out.println("Count: " + count);// 清除缓存df.unpersist();spark.stop();}
}

3. UDF (用户自定义函数)

你可以通过 UDF 在 Spark SQL 中使用自定义的 Java 函数。以下是如何在 Java 中注册并使用 UDF 的示例：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;public class SparkUDFExample {public static void main(String[] args) {SparkSession spark = SparkSession.builder().appName("UDF Example").master("local[*]").getOrCreate();// 读取 JSON 文件为 DataFrameDataset<Row> df = spark.read().json("hdfs://path/to/json");// 注册 UDF 将字符串转换为大写spark.udf().register("toUpperCase", new UDF1<String, String>() {@Overridepublic String call(String s) {return s.toUpperCase();}}, DataTypes.StringType);// 使用 UDF 在 SQL 查询中调用df.createOrReplaceTempView("people");Dataset<Row> result = spark.sql("SELECT name, toUpperCase(name) AS name_upper FROM people");// 显示结果result.show();spark.stop();}
}

4. 持久化 (Persistence) 和自定义存储级别

在 Spark 中，可以选择不同的持久化级别，以适应不同的场景。以下是使用 Java 代码实现自定义存储级别的示例：

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.sql.SparkSession;public class SparkPersistExample {public static void main(String[] args) {SparkSession spark = SparkSession.builder().appName("Persistence Example").master("local[*]").getOrCreate();JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());// 创建 RDDJavaRDD<String> rdd = sc.textFile("hdfs://path/to/text");// 使用自定义持久化级别，将数据存储在内存和磁盘中rdd.persist(StorageLevel.MEMORY_AND_DISK());// 对持久化的 RDD 进行操作long count = rdd.count();System.out.println("Count: " + count);// 清除持久化rdd.unpersist();spark.stop();}
}

5. Structured Streaming 实时流处理

Spark Structured Streaming 是用于实时流处理的高级 API。你可以从 Kafka 等数据源读取数据，并对其进行实时处理。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;public class SparkStreamingExample {public static void main(String[] args) throws StreamingQueryException {SparkSession spark = SparkSession.builder().appName("Structured Streaming Example").master("local[*]").getOrCreate();// 从 Kafka 中读取实时数据Dataset<Row> kafkaStream = spark.readStream().format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test_topic").load();// 将 Kafka 的消息转为字符串Dataset<Row> values = kafkaStream.selectExpr("CAST(value AS STRING)");// 开始流式查询，将结果输出到控制台StreamingQuery query = values.writeStream().outputMode("append").format("console").start();// 等待流式查询结束query.awaitTermination();}
}