如何使用JAVA Spark构建DataFrame？

本文共计409个文字，预计阅读时间需要2分钟。

讲述正传，接下来开始说正事。以前用Python和Scala操作Spark的时候，比较多的，毕业后写Python和Scala代码要简洁很多。今天一起来看看Java版本怎么创建DataFrame，代码写起来其实差异不大。

述说正传，接下来开始说正事。

以前用Python和Scala操作Spark的时候比较多，毕竟Python和Scala代码写起来要简洁很多。

今天一起来看看Java版本怎么创建DataFrame，代码写起来其实差不多，毕竟公用同一套API。测试数据可以参考我之前的文章。

先来总结下Spark的一般流程：

1，先创建Spark基础变量，spark，sc

2，加载数据，rdd.textFile，spark.read.csv/json等

3，数据处理，mapPartition, map，filter，reduce等一系列transformation操作

4，数据保存，saveAstextFile，或者其他DataFrame方法

祭出代码

package dev.java; import dev.utils.Utils; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.RowFactory; import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.types.StructType; import scala.Tuple2; import java.util.List; public class Spark1 { private static final String fileData = "seed"; private static final String fileSave = "result"; private static SparkSession spark = SparkSession.builder() .appName("Java-Spark") .master("local[*]") .config("spark.default.parallelism", 100) .config("spark.sql.shuffle.partitions", 100) .config("spark.driver.maxResultSize", "3g") .getOrCreate(); private static JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext()); public static void main(String[] args) { Utils.delete(fileSave); // t1(); } private static void t1() { JavaRDD<Row> rdd = sc.textFile(fileData) .map(v -> { String[] parts = v.split("\t"); return RowFactory.create(parts[0], Long.parseLong(parts[1])); }) .filter(v -> v.getLong(1) >= 10000) .sortBy(v -> v.getLong(1), false, 100) .coalesce(2); Dataset<Row> df = spark.createDataFrame(rdd, StructType.fromDDL("title string, qty long")); df.write().csv(fileSave); spark.stop(); } }

以上就是JAVA操作spark创建DataFrame的方法的详细内容，更多关于JAVA Spark 创建DataFrame的资料请关注易盾网络其它相关文章！